omnibias.jax.confluent_bank

0.5.0rc1Release candidatePython >=3.10Apache-2.0

JAX activation derivatives, functional network jets, direct contractions, and parameter-differentiable numerical models.

import omnibias.jax.confluent_bank

Module contract & mathematical conventions

Functional, preallocated collision bank with versioned host transitions.

Forward, retraction and parameter replacement are JIT-compatible. Proposals, checkpoint ingestion and acceptance run between optimizer steps on the host. The explicit parameter dictionary excludes integer/bool representation state.

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.

Module source ↗Source revision 9a2d73df0573Running reference examples →

CLASS

ConfluentPackBank

omnibias.jax.confluent_bank.ConfluentPackBankSource ↗

class ConfluentPackBank

Dataclass constructor fields; inherited fields, if any, are declared by the base class. A field(...) expression describes a default factory or field options, not a literal argument default.

The module contract describes the context for this operation. Its signature, explicit validation, return expressions, and implementation are recorded below.

Parameters

Name	Type & default	Contract
`centers`dataclass field	`Array`Required	See the operation contract and implementation below.
`weights`dataclass field	`Array`Required	See the operation contract and implementation below.
`scales`dataclass field	`Array`Required	See the operation contract and implementation below.
`rho`dataclass field	`Array`Required	See the operation contract and implementation below.
`odd_moments`dataclass field	`Array`Required	See the operation contract and implementation below.
`directions`dataclass field	`Array`Required	See the operation contract and implementation below.
`direction_bias`dataclass field	`Array`Required	See the operation contract and implementation below.
`moment_coefficients`dataclass field	`Array`Required	See the operation contract and implementation below.
`orders`dataclass field	`Array`Required	See the operation contract and implementation below.
`ages`dataclass field	`Array`Required	See the operation contract and implementation below.
`active`dataclass field	`Array`Required	See the operation contract and implementation below.
`modes`dataclass field	`Array`Required	See the operation contract and implementation below.
`slot_ids`dataclass field	`Array`Required	See the operation contract and implementation below.
`transition_version`dataclass field	`Array`Required	See the operation contract and implementation below.
`max_order`dataclass field	`int`Required	See the operation contract and implementation below.
`base`dataclass field	`str`Required	See the operation contract and implementation below.

Fields & methods

centersfield

centers: Array

Declared Array field on ConfluentPackBank. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

weightsfield

weights: Array

Declared Array field on ConfluentPackBank. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

scalesfield

scales: Array

Declared Array field on ConfluentPackBank. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

rhofield

rho: Array

Declared Array field on ConfluentPackBank. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

odd_momentsfield

odd_moments: Array

Declared Array field on ConfluentPackBank. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

directionsfield

directions: Array

Declared Array field on ConfluentPackBank. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

direction_biasfield

direction_bias: Array

Declared Array field on ConfluentPackBank. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

moment_coefficientsfield

moment_coefficients: Array

Declared Array field on ConfluentPackBank. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

ordersfield

orders: Array

Declared Array field on ConfluentPackBank. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

agesfield

ages: Array

Declared Array field on ConfluentPackBank. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

activefield

active: Array

Declared Array field on ConfluentPackBank. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

modesfield

modes: Array

Declared Array field on ConfluentPackBank. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

slot_idsfield

slot_ids: Array

Declared Array field on ConfluentPackBank. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

transition_versionfield

transition_version: Array

Declared Array field on ConfluentPackBank. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

max_orderfield

max_order: int

Declared int field on ConfluentPackBank. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

basefield

base: str

Declared str field on ConfluentPackBank. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

tree_flattenmethod

tree_flatten(self) -> tuple[tuple[Array, ...], tuple[int, str]]

See the class contract above and the exact method implementation below.

Returns: tuple[tuple[Array, ...], tuple[int, str]]

Method implementation

    def tree_flatten(self) -> tuple[tuple[Array, ...], tuple[int, str]]:
        return tuple(getattr(self, name) for name in STATE_NAMES), (self.max_order, self.base)

View source ↗

tree_unflattenmethod

tree_unflatten(cls, aux: tuple[int, str], children: tuple[Array, ...]) -> ConfluentPackBank

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
`aux`positional-or-keyword	`tuple[int, str]`Required	See the operation contract and implementation below.
`children`positional-or-keyword	`tuple[Array, ...]`Required	See the operation contract and implementation below.

Returns: ConfluentPackBank

Method implementation

    def tree_unflatten(cls, aux: tuple[int, str], children: tuple[Array, ...]) -> ConfluentPackBank:
        values: dict[str, Any] = dict(zip(STATE_NAMES, children, strict=True))
        return cls(**values, max_order=aux[0], base=aux[1])

View source ↗

createmethod

create(cls, packs: Sequence[RefinedPack], *, max_packs: int=16, max_order: int=8, base: str='sigmoid', dtype: Any=None) -> ConfluentPackBank

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
`packs`positional-or-keyword	`Sequence[RefinedPack]`Required	See the operation contract and implementation below.
`max_packs`keyword-only	`int`Default: 16	See the operation contract and implementation below.
`max_order`keyword-only	`int`Default: 8	See the operation contract and implementation below.
`base`keyword-only	`str`Default: 'sigmoid'	See the operation contract and implementation below.
`dtype`keyword-only	`Any`Default: None	See the operation contract and implementation below.

Returns: ConfluentPackBank

Method implementation

    def create(cls, packs: Sequence[RefinedPack], *, max_packs: int = 16, max_order: int = 8,
               base: str = "sigmoid", dtype: Any = None) -> ConfluentPackBank:
        return init_confluent_bank(packs, max_packs=max_packs, max_order=max_order, base=base, dtype=dtype)

View source ↗

max_packsproperty

max_packs(self) -> int

See the class contract above and the exact method implementation below.

Returns: int

Method implementation

    def max_packs(self) -> int:
        return int(self.centers.shape[0])

View source ↗

parametersproperty

parameters(self) -> dict[str, Array]

See the class contract above and the exact method implementation below.

Returns: dict[str, Array]

Method implementation

    def parameters(self) -> dict[str, Array]:
        return {name: getattr(self, name) for name in PARAMETER_NAMES}

View source ↗

with_parametersmethod

with_parameters(self, parameters: Mapping[str, Array]) -> ConfluentPackBank

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
`parameters`positional-or-keyword	`Mapping[str, Array]`Required	See the operation contract and implementation below.

Returns: ConfluentPackBank

Validation & failure cases

When set(parameters) != set(PARAMETER_NAMES), raises ValueError('parameters must contain exactly the trainable bank blocks').
When value.shape != getattr(self, name).shape, raises ValueError(f'parameter shape changed for {name}').

Method implementation

    def with_parameters(self, parameters: Mapping[str, Array]) -> ConfluentPackBank:
        if set(parameters) != set(PARAMETER_NAMES):
            raise ValueError("parameters must contain exactly the trainable bank blocks")
        for name, value in parameters.items():
            if value.shape != getattr(self, name).shape:
                raise ValueError(f"parameter shape changed for {name}")
        return replace(self, **dict(cast(Mapping[str, Any], parameters)))

View source ↗

forwardmethod

forward(self, z: Array) -> Array

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
`z`positional-or-keyword	`Array`Required	See the operation contract and implementation below.

Returns: Array

Method implementation

    def forward(self, z: Array) -> Array:
        return bank_forward(self, z)

View source ↗

slot_termsmethod

slot_terms(self, z: Array) -> Array

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
`z`positional-or-keyword	`Array`Required	See the operation contract and implementation below.

Returns: Array

Method implementation

    def slot_terms(self, z: Array) -> Array:
        return bank_slot_terms(self, z)

View source ↗

pack_term_normsmethod

pack_term_norms(self, z: Array) -> list[float]

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
`z`positional-or-keyword	`Array`Required	See the operation contract and implementation below.

Returns: list[float]

Method implementation

    def pack_term_norms(self, z: Array) -> list[float]:
        terms = self.slot_terms(z)
        return [float(jnp.sqrt(jnp.sum(terms[i] ** 2))) for i in range(self.max_packs) if bool(self.active[i])]

View source ↗

retractmethod

retract(self) -> ConfluentPackBank

See the class contract above and the exact method implementation below.

Returns: ConfluentPackBank

Method implementation

    def retract(self) -> ConfluentPackBank:
        return replace(self, rho=retract_spread(self.rho))

View source ↗

portable_snapshotmethod

portable_snapshot(self) -> dict[str, Any]

See the class contract above and the exact method implementation below.

Returns: dict[str, Any]

Method implementation

    def portable_snapshot(self) -> dict[str, Any]:
        snapshot = {name: {"shape": list(value.shape), "dtype": str(value.dtype),
                       "values": value.reshape(-1).tolist()}
                for name in STATE_NAMES for value in (getattr(self, name),)}
        snapshot["__configuration__"] = {"activation": self.base, "max_order": self.max_order,
                                           "max_packs": self.max_packs, "schema_version": 1,
                                           "series_terms": 6, "series_radius": 0.01}
        return snapshot

View source ↗

checkpointmethod

checkpoint(self) -> dict[str, Any]

See the class contract above and the exact method implementation below.

Returns: dict[str, Any]

Method implementation

    def checkpoint(self) -> dict[str, Any]:
        return {"schema_version": 1, "base": self.base, "max_order": self.max_order,
                "state": self.portable_snapshot()}

View source ↗

propose_clustermethod

propose_cluster(self, slots: tuple[int, ...], *, order: int, error_budget: float) -> TransitionProposal

Common-weight ordinary cluster -> bounded truncated moment expansion.

Parameters

Name	Type & default	Contract
`slots`positional-or-keyword	`tuple[int, ...]`Required	See the operation contract and implementation below.
`order`keyword-only	`int`Required	See the operation contract and implementation below.
`error_budget`keyword-only	`float`Required	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)), raises ValueError('valid distinct slots and an order within the bank budget required').
When scale == 0, raises ValueError('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])) != scale), raises ValueError('cluster requires ordinary order-zero atoms with a common weight').

Method 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]))
        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])) != scale):
                raise ValueError('cluster requires ordinary order-zero atoms with a common weight')
        centers = tuple(Fraction(float(self.centers[i])) 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]) 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)

View source ↗

propose_pairmethod

propose_pair(self, first: int, second: int, *, error_budget: float=0.0) -> TransitionProposal

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
`first`positional-or-keyword	`int`Required	See the operation contract and implementation below.
`second`positional-or-keyword	`int`Required	See the operation contract and implementation below.
`error_budget`keyword-only	`float`Default: 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))), raises ValueError('two distinct live slots required').
When not bool(self.active[i]) or int(self.modes[i]) != 0 or int(self.orders[i]) != 0, raises ValueError('pair proposal requires ordinary order-zero live slots').
When scale != Fraction(float(self.scales[second])) or scale == 0, raises ValueError('bias confluence requires identical nonzero weights').

Method implementation

    def propose_pair(self, first: int, second: int, *, error_budget: float = 0.0) -> TransitionProposal:
        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]))
        if scale != Fraction(float(self.scales[second])) or scale == 0:
            raise ValueError("bias confluence requires identical nonzero weights")
        b = [-scale * Fraction(float(self.centers[i])) for i in (first, second)]
        a = [Fraction(float(self.weights[i])) for i in (first, second)]
        half = 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, half * half), ("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(name, i, float(v)) for name, i, v in values),
                                  "duplicate_merge" if half == 0 else "centered_pair",
                                  "exact_real_coordinate_identity_with_recorded_float_operands", error_budget)

View source ↗

propose_derivativemethod

propose_derivative(self, slot: int, *, error_budget: float) -> TransitionProposal

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
`slot`positional-or-keyword	`int`Required	See the operation contract and implementation below.
`error_budget`keyword-only	`float`Required	See the operation contract and implementation below.

Returns: TransitionProposal

Validation & failure cases

When not bool(self.active[slot]) or int(self.modes[slot]) != 1, raises ValueError('a live centered-pair slot is required').

Method implementation

    def propose_derivative(self, slot: int, *, error_budget: float) -> TransitionProposal:
        self._check_slot(slot)
        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)

View source ↗

propose_remove_zeromethod

propose_remove_zero(self, slot: int) -> TransitionProposal

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
`slot`positional-or-keyword	`int`Required	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]) != 0, raises ValueError('an exactly zero ordinary output coefficient is required').

Method implementation

    def propose_remove_zero(self, slot: int) -> TransitionProposal:
        self._check_slot(slot)
        if not bool(self.active[slot]) or int(self.modes[slot]) != 0 or float(self.weights[slot]) != 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")

View source ↗

propose_birthmethod

propose_birth(self, slot: int, pack: RefinedPack) -> TransitionProposal

Birth/reuse an inactive slot with exactly zero output and fresh moments.

Parameters

Name	Type & default	Contract
`slot`positional-or-keyword	`int`Required	See the operation contract and implementation below.
`pack`positional-or-keyword	`RefinedPack`Required	See the operation contract and implementation below.

Returns: TransitionProposal

Validation & failure cases

When bool(self.active[slot]), raises ValueError('birth requires an allocated inactive slot').
When pack.weight != 0 or pack.order > self.max_order, raises ValueError('birth requires zero output weight and an order within budget').

Method implementation

    def propose_birth(self, slot: int, pack: RefinedPack) -> TransitionProposal:
        """Birth/reuse an inactive slot with exactly zero output and fresh moments."""
        self._check_slot(slot)
        if 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")

View source ↗

Read the implementation

class ConfluentPackBank:
    centers: Array
    weights: Array
    scales: Array
    rho: Array
    odd_moments: Array
    directions: Array
    direction_bias: Array
    moment_coefficients: Array
    orders: Array
    ages: Array
    active: Array
    modes: Array
    slot_ids: Array
    transition_version: Array
    max_order: int
    base: str

    def tree_flatten(self) -> tuple[tuple[Array, ...], tuple[int, str]]:
        return tuple(getattr(self, name) for name in STATE_NAMES), (self.max_order, self.base)

    @classmethod
    def tree_unflatten(cls, aux: tuple[int, str], children: tuple[Array, ...]) -> ConfluentPackBank:
        values: dict[str, Any] = dict(zip(STATE_NAMES, children, strict=True))
        return cls(**values, max_order=aux[0], base=aux[1])

    @classmethod
    def create(cls, packs: Sequence[RefinedPack], *, max_packs: int = 16, max_order: int = 8,
               base: str = "sigmoid", dtype: Any = None) -> ConfluentPackBank:
        return init_confluent_bank(packs, max_packs=max_packs, max_order=max_order, base=base, dtype=dtype)

    @property
    def max_packs(self) -> int:
        return int(self.centers.shape[0])

    @property
    def parameters(self) -> dict[str, Array]:
        return {name: getattr(self, name) for name in PARAMETER_NAMES}

    def with_parameters(self, parameters: Mapping[str, Array]) -> ConfluentPackBank:
        if set(parameters) != set(PARAMETER_NAMES):
            raise ValueError("parameters must contain exactly the trainable bank blocks")
        for name, value in parameters.items():
            if value.shape != getattr(self, name).shape:
                raise ValueError(f"parameter shape changed for {name}")
        return replace(self, **dict(cast(Mapping[str, Any], parameters)))

    def forward(self, z: Array) -> Array:
        return bank_forward(self, z)

    def slot_terms(self, z: Array) -> Array:
        return bank_slot_terms(self, z)

    def pack_term_norms(self, z: Array) -> list[float]:
        terms = self.slot_terms(z)
        return [float(jnp.sqrt(jnp.sum(terms[i] ** 2))) for i in range(self.max_packs) if bool(self.active[i])]

    def retract(self) -> ConfluentPackBank:
        return replace(self, rho=retract_spread(self.rho))

    def portable_snapshot(self) -> dict[str, Any]:
        snapshot = {name: {"shape": list(value.shape), "dtype": str(value.dtype),
                       "values": value.reshape(-1).tolist()}
                for name in STATE_NAMES for value in (getattr(self, name),)}
        snapshot["__configuration__"] = {"activation": self.base, "max_order": self.max_order,
                                           "max_packs": self.max_packs, "schema_version": 1,
                                           "series_terms": 6, "series_radius": 0.01}
        return snapshot

    def checkpoint(self) -> dict[str, Any]:
        return {"schema_version": 1, "base": self.base, "max_order": self.max_order,
                "state": self.portable_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]))
        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])) != scale):
                raise ValueError('cluster requires ordinary order-zero atoms with a common weight')
        centers = tuple(Fraction(float(self.centers[i])) 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]) 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:
        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]))
        if scale != Fraction(float(self.scales[second])) or scale == 0:
            raise ValueError("bias confluence requires identical nonzero weights")
        b = [-scale * Fraction(float(self.centers[i])) for i in (first, second)]
        a = [Fraction(float(self.weights[i])) for i in (first, second)]
        half = 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, half * half), ("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(name, i, float(v)) for name, i, v in values),
                                  "duplicate_merge" if half == 0 else "centered_pair",
                                  "exact_real_coordinate_identity_with_recorded_float_operands", error_budget)

    def propose_derivative(self, slot: int, *, error_budget: float) -> TransitionProposal:
        self._check_slot(slot)
        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:
        self._check_slot(slot)
        if not bool(self.active[slot]) or int(self.modes[slot]) != 0 or float(self.weights[slot]) != 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 _check_slot(self, slot: int) -> None:
        if slot < 0 or slot >= self.max_packs:
            raise ValueError("slot is outside the bank")

    def propose_birth(self, slot: int, pack: RefinedPack) -> TransitionProposal:
        """Birth/reuse an inactive slot with exactly zero output and fresh moments."""
        self._check_slot(slot)
        if 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")

CLASS

ExactTransportAdapter

omnibias.jax.confluent_bank.ExactTransportAdapterSource ↗

class ExactTransportAdapter(Protocol[StateT])

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, state: StateT, bank: ConfluentPackBank, permutation: tuple[int, ...] | None, sign_slots: tuple[int, ...]) -> StateT

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
`state`positional-or-keyword	`StateT`Required	See the operation contract and implementation below.
`bank`positional-or-keyword	`ConfluentPackBank`Required	See the operation contract and implementation below.
`permutation`positional-or-keyword	`tuple[int, ...] \| None`Required	See the operation contract and implementation below.
`sign_slots`positional-or-keyword	`tuple[int, ...]`Required	See the operation contract and implementation below.

Returns: StateT

Method implementation

    def __call__(self, state: StateT, bank: ConfluentPackBank,
                 permutation: tuple[int, ...] | None, sign_slots: tuple[int, ...]) -> StateT: ...

View source ↗

Read the implementation

class ExactTransportAdapter(Protocol[StateT]):
    def __call__(self, state: StateT, bank: ConfluentPackBank,
                 permutation: tuple[int, ...] | None, sign_slots: tuple[int, ...]) -> StateT: ...

CLASS

NamedMomentState

omnibias.jax.confluent_bank.NamedMomentStateSource ↗

class NamedMomentState

Dataclass constructor fields; inherited fields, if any, are declared by the base class. A field(...) expression describes a default factory or field options, not a literal argument default.

Explicit elementwise optimizer moments and a preserved shared step count.

Each moment dictionary has exactly the bank's parameter names/shapes. This represents Adam's first/second moments or an SGD momentum without guessing the structure of an arbitrary optimizer's state. Adapt other states through OptimizerStateAdapter.

Parameters

Name	Type & default	Contract
`step`dataclass field	`Array`Required	See the operation contract and implementation below.
`moments`dataclass field	`dict[str, dict[str, Array]]`Required	See the operation contract and implementation below.

Fields & methods

stepfield

step: Array

Declared Array field on NamedMomentState. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

momentsfield

moments: dict[str, dict[str, Array]]

Declared dict[str, dict[str, Array]] field on NamedMomentState. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

tree_flattenmethod

tree_flatten(self) -> tuple[tuple[Array, dict[str, dict[str, Array]]], None]

See the class contract above and the exact method implementation below.

Returns: tuple[tuple[Array, dict[str, dict[str, Array]]], None]

Method implementation

    def tree_flatten(self) -> tuple[tuple[Array, dict[str, dict[str, Array]]], None]:
        return (self.step, self.moments), None

View source ↗

tree_unflattenmethod

tree_unflatten(cls, aux: None, children: tuple[Array, dict[str, dict[str, Array]]]) -> NamedMomentState

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
`aux`positional-or-keyword	`None`Required	See the operation contract and implementation below.
`children`positional-or-keyword	`tuple[Array, dict[str, dict[str, Array]]]`Required	See the operation contract and implementation below.

Returns: NamedMomentState

Method implementation

    def tree_unflatten(cls, aux: None, children: tuple[Array, dict[str, dict[str, Array]]]) -> NamedMomentState:
        return cls(*children)

View source ↗

Read the implementation

class NamedMomentState:
    """Explicit elementwise optimizer moments and a preserved shared step count.

    Each moment dictionary has exactly the bank's parameter names/shapes. This
    represents Adam's first/second moments or an SGD momentum without guessing
    the structure of an arbitrary optimizer's state. Adapt other states through
    ``OptimizerStateAdapter``.
    """

    step: Array
    moments: dict[str, dict[str, Array]]

    def tree_flatten(self) -> tuple[tuple[Array, dict[str, dict[str, Array]]], None]:
        return (self.step, self.moments), None

    @classmethod
    def tree_unflatten(cls, aux: None, children: tuple[Array, dict[str, dict[str, Array]]]) -> NamedMomentState:
        return cls(*children)

CLASS

OptimizerStateAdapter

omnibias.jax.confluent_bank.OptimizerStateAdapterSource ↗

class OptimizerStateAdapter(Protocol[StateT])

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, state: StateT, bank: ConfluentPackBank, slots: tuple[int, ...]) -> StateT

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
`state`positional-or-keyword	`StateT`Required	See the operation contract and implementation below.
`bank`positional-or-keyword	`ConfluentPackBank`Required	See the operation contract and implementation below.
`slots`positional-or-keyword	`tuple[int, ...]`Required	See the operation contract and implementation below.

Returns: StateT

Method implementation

    def __call__(self, state: StateT, bank: ConfluentPackBank,
                 slots: tuple[int, ...]) -> StateT: ...

View source ↗

Read the implementation

class OptimizerStateAdapter(Protocol[StateT]):
    def __call__(self, state: StateT, bank: ConfluentPackBank,
                 slots: tuple[int, ...]) -> StateT: ...

CONSTANT

PARAMETER_NAMES

omnibias.jax.confluent_bank.PARAMETER_NAMESSource ↗

PARAMETER_NAMES = ('centers', 'weights', 'scales', 'rho', 'odd_moments', 'directions', 'direction_bias', 'moment_coefficients')

This exported constant is defined by the expression above. Its exact value and annotation come from the pinned source; use it with the module contract.

Read the implementation

PARAMETER_NAMES = ("centers", "weights", "scales", "rho", "odd_moments", "directions",
                   "direction_bias", "moment_coefficients")

CONSTANT

STATE_NAMES

omnibias.jax.confluent_bank.STATE_NAMESSource ↗

STATE_NAMES = PARAMETER_NAMES + ('orders', 'ages', 'active', 'modes', 'slot_ids', 'transition_version')

This exported constant is defined by the expression above. Its exact value and annotation come from the pinned source; use it with the module contract.

Read the implementation

STATE_NAMES = PARAMETER_NAMES + ("orders", "ages", "active", "modes", "slot_ids", "transition_version")

FUNCTION

bank_forward

omnibias.jax.confluent_bank.bank_forwardSource ↗

bank_forward(bank: ConfluentPackBank, z: Array) -> Array

The module contract describes the context for this operation. Its signature, explicit validation, return expressions, and implementation are recorded below.

Parameters

Name	Type & default	Contract
`bank`positional-or-keyword	`ConfluentPackBank`Required	See the operation contract and implementation below.
`z`positional-or-keyword	`Array`Required	See the operation contract and implementation below.

ReturnsArray

Read the implementation

def bank_forward(bank: ConfluentPackBank, z: Array) -> Array:
    terms = bank_slot_terms(bank, z)
    result = jnp.zeros_like(z)
    for i in range(bank.max_packs):
        result = result + terms[i]
    return result

FUNCTION

bank_from_checkpoint

omnibias.jax.confluent_bank.bank_from_checkpointSource ↗

bank_from_checkpoint(checkpoint: Mapping[str, Any]) -> ConfluentPackBank

Validate a portable checkpoint. This host API never silently truncates dtype.

Parameters

Name	Type & default	Contract
`checkpoint`positional-or-keyword	`Mapping[str, Any]`Required	See the operation contract and implementation below.

ReturnsConfluentPackBank

Explicit validation & failure cases

When checkpoint.get('schema_version') != 1, raises ValueError('unsupported checkpoint schema').
When set(state) != set(STATE_NAMES) | {'__configuration__'}, raises ValueError('checkpoint contains missing or unknown state blocks').
When state['__configuration__'] != fresh.portable_snapshot()['__configuration__'], raises ValueError('checkpoint configuration disagrees with its bound state').
When value.shape != getattr(fresh, name).shape or str(value.dtype) != item['dtype'], raises ValueError(f'checkpoint shape/dtype not supported for {name}').
When name in PARAMETER_NAMES and (not jnp.issubdtype(value.dtype, jnp.floating) or value.dtype != fresh.centers.dtype), raises ValueError('checkpoint trainables must share a real floating dtype').
When name == 'active' and value.dtype != jnp.bool_, raises ValueError('checkpoint activity flags must be boolean').
When name not in PARAMETER_NAMES + ('active',) and (not jnp.issubdtype(value.dtype, jnp.integer)), raises ValueError('checkpoint representation metadata must be integer').
When not bool(jnp.all(jnp.isfinite(value))), raises ValueError('checkpoint state must be finite').
When not bool(jnp.all((candidate.modes >= 0) & (candidate.modes <= 2))) or not bool(jnp.all((candidate.orders >= 0) & (candidate.orders <= candidate.max_order))) or (not bool(jnp.all(candidate.rho >= 0))) or (not bool(jnp.all(candidate.slot_ids >= 0))) or (not bool(jnp.all(candidate.ages >= 0))) or (int(candidate.transition_version) < 0) or (len(set(candidate.slot_ids.tolist())) != capacity), raises ValueError('checkpoint representation state is invalid').

These are explicit source guards. Backend operations may also raise for invalid shapes, dtypes, or unsupported devices.

Regression example · checkpoint roundtrip configuration binding and legacy migration

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.

Full file, imports & fixtures ↗

def test_checkpoint_roundtrip_configuration_binding_and_legacy_migration() -> None:
    torch_bank, bank = pair_banks()
    restored = jb.bank_from_checkpoint(bank.checkpoint())
    assert restored.portable_snapshot() == bank.portable_snapshot()
    bad = deepcopy(bank.checkpoint())
    bad["base"] = "tanh"
    with pytest.raises(ValueError, match="configuration"):
        jb.bank_from_checkpoint(bad)
    bad = deepcopy(bank.checkpoint())
    bad["state"]["modes"]["values"][0] = 8
    with pytest.raises(ValueError, match="representation"):
        jb.bank_from_checkpoint(bad)
    bad = deepcopy(bank.checkpoint())
    bad["state"]["active"]["dtype"] = "float64"
    with pytest.raises(ValueError, match="boolean"):
        jb.bank_from_checkpoint(bad)
    changed_activation = replace(bank, base="tanh")
    with pytest.raises(ValueError, match="stale"):
        jb.commit_transition(changed_activation, bank.propose_pair(0, 1), moments(bank),
                              between_steps=True, acceptance=lambda _b, _p: True)
    packs = [RefinedPack(0, 0.1, 0.7), RefinedPack(1, 0.3, 0.2)]
    legacy = init_pack_bank(packs, max_packs=4, max_order=3, base="sigmoid")
    migrated = jb.migrate_legacy(legacy)
    x = jnp.array([0.1, 0.2])
    np.testing.assert_allclose(np.asarray(migrated.forward(x)), np.asarray(legacy_forward(legacy, x)), atol=1e-16)
    torch_legacy = AdaptivePackBank(packs, max_packs=4, max_order=3, base="sigmoid", dtype=torch.float64)
    torch_bank.load_state_dict(torch_legacy.state_dict())
    assert not bool(torch_bank.modes.any())
    np.testing.assert_allclose(torch_bank(torch.tensor(np.asarray(x))).detach().numpy(),
                               np.asarray(migrated.forward(x)), atol=1e-16)

python -m pytest tests/test_confluent_bank_parity.py -k test_checkpoint_roundtrip_configuration_binding_and_legacy_migration -q

Read the implementation

def bank_from_checkpoint(checkpoint: Mapping[str, Any]) -> ConfluentPackBank:
    """Validate a portable checkpoint. This host API never silently truncates dtype."""
    if checkpoint.get("schema_version") != 1:
        raise ValueError("unsupported checkpoint schema")
    state = checkpoint["state"]
    if set(state) != set(STATE_NAMES) | {"__configuration__"}:
        raise ValueError("checkpoint contains missing or unknown state blocks")
    capacity = state["centers"]["shape"][0]
    fresh = init_confluent_bank((), max_packs=capacity, max_order=checkpoint["max_order"],
                               base=checkpoint["base"], dtype=state["centers"]["dtype"])
    if state["__configuration__"] != fresh.portable_snapshot()["__configuration__"]:
        raise ValueError("checkpoint configuration disagrees with its bound state")
    values: dict[str, Any] = {}
    for name in STATE_NAMES:
        item = state[name]
        value = jnp.asarray(item["values"], dtype=item["dtype"]).reshape(tuple(item["shape"]))
        if value.shape != getattr(fresh, name).shape or str(value.dtype) != item["dtype"]:
            raise ValueError(f"checkpoint shape/dtype not supported for {name}")
        if name in PARAMETER_NAMES and (not jnp.issubdtype(value.dtype, jnp.floating)
                                         or value.dtype != fresh.centers.dtype):
            raise ValueError("checkpoint trainables must share a real floating dtype")
        if name == "active" and value.dtype != jnp.bool_:
            raise ValueError("checkpoint activity flags must be boolean")
        if name not in PARAMETER_NAMES + ("active",) and not jnp.issubdtype(value.dtype, jnp.integer):
            raise ValueError("checkpoint representation metadata must be integer")
        if not bool(jnp.all(jnp.isfinite(value))):
            raise ValueError("checkpoint state must be finite")
        values[name] = value
    candidate = replace(fresh, **values)
    if (not bool(jnp.all((candidate.modes >= 0) & (candidate.modes <= 2)))
            or not bool(jnp.all((candidate.orders >= 0) & (candidate.orders <= candidate.max_order)))
            or not bool(jnp.all(candidate.rho >= 0))
            or not bool(jnp.all(candidate.slot_ids >= 0))
            or not bool(jnp.all(candidate.ages >= 0))
            or int(candidate.transition_version) < 0
            or len(set(candidate.slot_ids.tolist())) != capacity):
        raise ValueError("checkpoint representation state is invalid")
    return candidate

FUNCTION

bank_slot_terms

omnibias.jax.confluent_bank.bank_slot_termsSource ↗

bank_slot_terms(bank: ConfluentPackBank, z: Array) -> Array

One shared registered tower per slot, reused by all three mode branches.

Parameters

Name	Type & default	Contract
`bank`positional-or-keyword	`ConfluentPackBank`Required	See the operation contract and implementation below.
`z`positional-or-keyword	`Array`Required	See the operation contract and implementation below.

ReturnsArray

Read the implementation

def bank_slot_terms(bank: ConfluentPackBank, z: Array) -> Array:
    """One shared registered tower per slot, reused by all three mode branches."""
    spec = get_activation(bank.base)
    terms = []
    for i in range(bank.max_packs):
        argument = bank.scales[i] * (z - bank.centers[i])
        tower = _sigma_tower(spec, argument, max(bank.max_order, 11))
        ordinary = jnp.zeros_like(z)
        for order in range(bank.max_order + 1):
            value = bank.weights[i] * (bank.scales[i] ** order) * tower[order]
            ordinary = ordinary + jnp.where(bank.orders[i] == order, value, jnp.zeros_like(z))
        pair = collision_atom(argument, bank.rho[i], bank.directions[i] * z + bank.direction_bias[i],
                              bank.weights[i], bank.odd_moments[i], activation=bank.base, tower=tower)
        moments = moment_atom(argument, bank.moment_coefficients[i], activation=bank.base, tower=tower)
        value = jnp.where(bank.modes[i] == 0, ordinary, jnp.where(bank.modes[i] == 1, pair, moments))
        terms.append(jnp.where(bank.active[i], value, jnp.zeros_like(z)))
    return jnp.stack(terms)

FUNCTION

commit_transition

omnibias.jax.confluent_bank.commit_transitionSource ↗

commit_transition(bank: ConfluentPackBank, proposal: TransitionProposal, optimizer_state: StateT, *, between_steps: bool, acceptance: Callable[[ConfluentPackBank, TransitionProposal], bool], postcondition: Callable[[ConfluentPackBank], bool] | None=None, state_adapter: OptimizerStateAdapter[StateT] | None=None) -> tuple[ConfluentPackBank, StateT, bool]

Functional host transaction. Rejection returns both original objects.

The acceptance callback checks the actual requested error budget before any state is constructed. A failed postcondition discards both candidate model and adapted optimizer state. No callback here is itself a proof.

Parameters

Name	Type & default	Contract
`bank`positional-or-keyword	`ConfluentPackBank`Required	See the operation contract and implementation below.
`proposal`positional-or-keyword	`TransitionProposal`Required	See the operation contract and implementation below.
`optimizer_state`positional-or-keyword	`StateT`Required	See the operation contract and implementation below.
`between_steps`keyword-only	`bool`Required	See the operation contract and implementation below.
`acceptance`keyword-only	`Callable[[ConfluentPackBank, TransitionProposal], bool]`Required	See the operation contract and implementation below.
`postcondition`keyword-only	`Callable[[ConfluentPackBank], bool] \| None`Default: None	See the operation contract and implementation below.
`state_adapter`keyword-only	`OptimizerStateAdapter[StateT] \| None`Default: None	See the operation contract and implementation below.

Returnstuple[ConfluentPackBank, StateT, bool]

Explicit validation & failure cases

When not between_steps, raises RuntimeError('commit only between optimizer steps').
When proposal.source_version != int(bank.transition_version) or proposal.source_digest != snapshot_digest(bank.portable_snapshot()), raises ValueError('stale transition proposal').
When any((i not in ids for i in proposal.slot_ids)), raises ValueError('proposal contains unknown stable slot IDs').
When update.name not in STATE_NAMES or update.name in ('slot_ids', 'transition_version'), raises ValueError('proposal may only update representation tensors').
When tensor.ndim == 0 or update.index not in slots, raises ValueError('update must refer to an affected slot').
When value.shape != tensor[update.index].shape, raises ValueError('slot update shape does not match the preallocated row').
When not isinstance(optimizer_state, NamedMomentState), raises TypeError('unknown optimizer state requires an explicit OptimizerStateAdapter').

These are explicit source guards. Backend operations may also raise for invalid shapes, dtypes, or unsupported devices.

Regression example · jit parameter gradient retraction and moment mode

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.

Full file, imports & fixtures ↗

def test_jit_parameter_gradient_retraction_and_moment_mode() -> None:
    _, bank = pair_banks()
    bank, _, _ = jb.commit_transition(bank, bank.propose_pair(0, 1), moments(bank), between_steps=True,
                                      acceptance=lambda _b, _p: True)
    x = jnp.asarray([-0.3, 0.2])
    compiled = jax.jit(lambda state, z: state.forward(z))
    np.testing.assert_array_equal(np.asarray(compiled(bank, x)), np.asarray(bank.forward(x)))
    def objective(parameters):
        return jnp.sum(bank.with_parameters(parameters).forward(x) ** 2)
    gradients = jax.jit(jax.grad(objective))(bank.parameters)
    assert all(bool(jnp.all(jnp.isfinite(value))) for value in gradients.values())
    assert float(jnp.abs(gradients["odd_moments"][0])) > 0
    boundary = replace(bank, rho=bank.rho.at[0].set(-0.5)).retract()
    assert float(boundary.rho[0]) == 0
    assert float(jax.grad(lambda r: jb.retract_spread(r))(jnp.asarray(0.0))) == 1
    moment_bank = replace(boundary, modes=boundary.modes.at[0].set(2),
                          moment_coefficients=boundary.moment_coefficients.at[0, :2].set(jnp.array([0.7, 0.2])))
    assert bool(jnp.all(jnp.isfinite(compiled(moment_bank, x))))

python -m pytest tests/test_confluent_bank_parity.py -k test_jit_parameter_gradient_retraction_and_moment_mode -q

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.

Full file, imports & fixtures ↗

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]) > 0

python -m pytest tests/test_confluent_bank_parity.py -k test_cluster_row_transition_and_error_gated_derivative -q

Read the implementation

def commit_transition(
    bank: ConfluentPackBank, proposal: TransitionProposal, optimizer_state: StateT,
    *, between_steps: bool, acceptance: Callable[[ConfluentPackBank, TransitionProposal], bool],
    postcondition: Callable[[ConfluentPackBank], bool] | None = None,
    state_adapter: OptimizerStateAdapter[StateT] | None = None,
) -> tuple[ConfluentPackBank, StateT, bool]:
    """Functional host transaction. Rejection returns both original objects.

    The acceptance callback checks the actual requested error budget before
    any state is constructed. A failed postcondition discards both candidate
    model and adapted optimizer state. No callback here is itself a proof.
    """
    if not between_steps:
        raise RuntimeError("commit only between optimizer steps")
    if (proposal.source_version != int(bank.transition_version)
            or proposal.source_digest != snapshot_digest(bank.portable_snapshot())):
        raise ValueError("stale transition proposal")
    ids = bank.slot_ids.tolist()
    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 bank, optimizer_state, False
    changed: dict[str, Any] = {}
    for update in proposal.updates:
        if update.name not in STATE_NAMES or update.name in ("slot_ids", "transition_version"):
            raise ValueError("proposal may only update representation tensors")
        tensor = changed.get(update.name, getattr(bank, update.name))
        if tensor.ndim == 0 or update.index not in slots:
            raise ValueError("update must refer to an affected slot")
        value = jnp.asarray(update.value, dtype=tensor.dtype)
        if value.shape != tensor[update.index].shape:
            raise ValueError("slot update shape does not match the preallocated row")
        changed[update.name] = tensor.at[update.index].set(value)
    candidate = replace(bank, **changed, transition_version=bank.transition_version + 1)
    if state_adapter is None:
        if not isinstance(optimizer_state, NamedMomentState):
            raise TypeError("unknown optimizer state requires an explicit OptimizerStateAdapter")
        candidate_state = cast(StateT, reset_slot_state(optimizer_state, candidate, slots))
    else:
        candidate_state = state_adapter(optimizer_state, candidate, slots)
    if postcondition is not None and not postcondition(candidate):
        return bank, optimizer_state, False
    return candidate, candidate_state, True

FUNCTION

detect_transitions

omnibias.jax.confluent_bank.detect_transitionsSource ↗

detect_transitions(bank: ConfluentPackBank, conditioning_score: Callable[[ConfluentPackBank], float], *, error_budget: float=0.0) -> tuple[TransitionProposal, ...]

The module contract describes the context for this operation. Its signature, explicit validation, return expressions, and implementation are recorded below.

Parameters

Name	Type & default	Contract
`bank`positional-or-keyword	`ConfluentPackBank`Required	See the operation contract and implementation below.
`conditioning_score`positional-or-keyword	`Callable[[ConfluentPackBank], float]`Required	See the operation contract and implementation below.
`error_budget`keyword-only	`float`Default: 0.0	See the operation contract and implementation below.

Returnstuple[TransitionProposal, ...]

Read the implementation

def detect_transitions(bank: ConfluentPackBank, conditioning_score: Callable[[ConfluentPackBank], float],
                       *, error_budget: float = 0.0) -> tuple[TransitionProposal, ...]:
    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
            state = NamedMomentState(jnp.asarray(0), {})
            candidate, _, accepted = commit_transition(bank, proposal, state, between_steps=True,
                                                       acceptance=lambda _bank, _proposal: True)
            if accepted:
                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

init_confluent_bank

omnibias.jax.confluent_bank.init_confluent_bankSource ↗

init_confluent_bank(packs: Sequence[RefinedPack], *, max_packs: int=16, max_order: int=8, base: str='sigmoid', dtype: Any=None) -> ConfluentPackBank

The module contract describes the context for this operation. Its signature, explicit validation, return expressions, and implementation are recorded below.

Parameters

Name	Type & default	Contract
`packs`positional-or-keyword	`Sequence[RefinedPack]`Required	See the operation contract and implementation below.
`max_packs`keyword-only	`int`Default: 16	See the operation contract and implementation below.
`max_order`keyword-only	`int`Default: 8	See the operation contract and implementation below.
`base`keyword-only	`str`Default: 'sigmoid'	See the operation contract and implementation below.
`dtype`keyword-only	`Any`Default: None	See the operation contract and implementation below.

ReturnsConfluentPackBank

Explicit validation & failure cases

When base not in ('sigmoid', 'tanh') or max_packs < 1 or max_order < 0, raises ValueError('positive capacity, nonnegative order, and sigmoid/tanh are required').
When len(live) > max_packs or any((pack.order > max_order for pack in live)), raises ValueError('initial packs exceed capacity or supported order').

These are explicit source guards. Backend operations may also raise for invalid shapes, dtypes, or unsupported devices.

Regression 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.

Full file, imports & fixtures ↗

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() == source

python -m pytest tests/test_confluent_bank_parity.py -k test_coupled_optimizer_requires_adapter_and_rolls_back -q

Regression example · jit parameter gradient retraction and moment mode

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.

Full file, imports & fixtures ↗

def test_jit_parameter_gradient_retraction_and_moment_mode() -> None:
    _, bank = pair_banks()
    bank, _, _ = jb.commit_transition(bank, bank.propose_pair(0, 1), moments(bank), between_steps=True,
                                      acceptance=lambda _b, _p: True)
    x = jnp.asarray([-0.3, 0.2])
    compiled = jax.jit(lambda state, z: state.forward(z))
    np.testing.assert_array_equal(np.asarray(compiled(bank, x)), np.asarray(bank.forward(x)))
    def objective(parameters):
        return jnp.sum(bank.with_parameters(parameters).forward(x) ** 2)
    gradients = jax.jit(jax.grad(objective))(bank.parameters)
    assert all(bool(jnp.all(jnp.isfinite(value))) for value in gradients.values())
    assert float(jnp.abs(gradients["odd_moments"][0])) > 0
    boundary = replace(bank, rho=bank.rho.at[0].set(-0.5)).retract()
    assert float(boundary.rho[0]) == 0
    assert float(jax.grad(lambda r: jb.retract_spread(r))(jnp.asarray(0.0))) == 1
    moment_bank = replace(boundary, modes=boundary.modes.at[0].set(2),
                          moment_coefficients=boundary.moment_coefficients.at[0, :2].set(jnp.array([0.7, 0.2])))
    assert bool(jnp.all(jnp.isfinite(compiled(moment_bank, x))))

python -m pytest tests/test_confluent_bank_parity.py -k test_jit_parameter_gradient_retraction_and_moment_mode -q

Read the implementation

def init_confluent_bank(packs: Sequence[RefinedPack], *, max_packs: int = 16, max_order: int = 8,
                        base: str = "sigmoid", dtype: Any = None) -> ConfluentPackBank:
    if base not in ("sigmoid", "tanh") or max_packs < 1 or max_order < 0:
        raise ValueError("positive capacity, nonnegative order, and sigmoid/tanh are required")
    live = tuple(packs)
    if len(live) > max_packs or any(pack.order > max_order for pack in live):
        raise ValueError("initial packs exceed capacity or supported order")
    centers, weights, scales = [0.0] * max_packs, [0.0] * max_packs, [1.0] * max_packs
    orders, ages, active = [0] * max_packs, [0] * max_packs, [False] * max_packs
    for i, pack in enumerate(live):
        centers[i], weights[i], scales[i] = pack.center, pack.weight, pack.scale
        orders[i], ages[i], active[i] = pack.order, pack.age, True
    center_array = jnp.asarray(centers, dtype=dtype)
    dt = center_array.dtype
    # Match the workspace's 64-bit integer snapshot schema when x64 is enabled.
    it = jnp.asarray(0).dtype
    return ConfluentPackBank(center_array, jnp.asarray(weights, dtype=dt), jnp.asarray(scales, dtype=dt),
                             jnp.zeros(max_packs, dtype=dt), jnp.zeros(max_packs, dtype=dt),
                             jnp.zeros(max_packs, dtype=dt), jnp.ones(max_packs, dtype=dt),
                             jnp.zeros((max_packs, max_order + 1), dtype=dt),
                             jnp.asarray(orders, dtype=it), jnp.asarray(ages, dtype=it),
                             jnp.asarray(active), jnp.zeros(max_packs, dtype=it),
                             jnp.arange(max_packs, dtype=it), jnp.asarray(0, dtype=it), max_order, base)

FUNCTION

migrate_legacy

omnibias.jax.confluent_bank.migrate_legacySource ↗

migrate_legacy(bank: AdaptivePackBank) -> ConfluentPackBank

Add ordinary-mode state to an existing functional bank without moving slots.

Parameters

Name	Type & default	Contract
`bank`positional-or-keyword	`AdaptivePackBank`Required	See the operation contract and implementation below.

ReturnsConfluentPackBank

Regression example · checkpoint roundtrip configuration binding and legacy migration

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.

Full file, imports & fixtures ↗

def test_checkpoint_roundtrip_configuration_binding_and_legacy_migration() -> None:
    torch_bank, bank = pair_banks()
    restored = jb.bank_from_checkpoint(bank.checkpoint())
    assert restored.portable_snapshot() == bank.portable_snapshot()
    bad = deepcopy(bank.checkpoint())
    bad["base"] = "tanh"
    with pytest.raises(ValueError, match="configuration"):
        jb.bank_from_checkpoint(bad)
    bad = deepcopy(bank.checkpoint())
    bad["state"]["modes"]["values"][0] = 8
    with pytest.raises(ValueError, match="representation"):
        jb.bank_from_checkpoint(bad)
    bad = deepcopy(bank.checkpoint())
    bad["state"]["active"]["dtype"] = "float64"
    with pytest.raises(ValueError, match="boolean"):
        jb.bank_from_checkpoint(bad)
    changed_activation = replace(bank, base="tanh")
    with pytest.raises(ValueError, match="stale"):
        jb.commit_transition(changed_activation, bank.propose_pair(0, 1), moments(bank),
                              between_steps=True, acceptance=lambda _b, _p: True)
    packs = [RefinedPack(0, 0.1, 0.7), RefinedPack(1, 0.3, 0.2)]
    legacy = init_pack_bank(packs, max_packs=4, max_order=3, base="sigmoid")
    migrated = jb.migrate_legacy(legacy)
    x = jnp.array([0.1, 0.2])
    np.testing.assert_allclose(np.asarray(migrated.forward(x)), np.asarray(legacy_forward(legacy, x)), atol=1e-16)
    torch_legacy = AdaptivePackBank(packs, max_packs=4, max_order=3, base="sigmoid", dtype=torch.float64)
    torch_bank.load_state_dict(torch_legacy.state_dict())
    assert not bool(torch_bank.modes.any())
    np.testing.assert_allclose(torch_bank(torch.tensor(np.asarray(x))).detach().numpy(),
                               np.asarray(migrated.forward(x)), atol=1e-16)

python -m pytest tests/test_confluent_bank_parity.py -k test_checkpoint_roundtrip_configuration_binding_and_legacy_migration -q

Read the implementation

def migrate_legacy(bank: AdaptivePackBank) -> ConfluentPackBank:
    """Add ordinary-mode state to an existing functional bank without moving slots."""
    fresh = init_confluent_bank((), max_packs=bank.max_packs, max_order=bank.max_order,
                               base=bank.base, dtype=bank.centers.dtype)
    return replace(fresh, centers=bank.centers, weights=bank.weights, scales=bank.scales,
                   orders=bank.orders, ages=bank.ages, active=bank.active)

FUNCTION

permute_bank

omnibias.jax.confluent_bank.permute_bankSource ↗

permute_bank(bank: ConfluentPackBank, permutation: Sequence[int], optimizer_state: StateT, *, between_steps: bool, state_adapter: ExactTransportAdapter[StateT] | None=None) -> tuple[ConfluentPackBank, StateT]

Move stable IDs, every state row and its moments through one exact permutation.

Parameters

Name	Type & default	Contract
`bank`positional-or-keyword	`ConfluentPackBank`Required	See the operation contract and implementation below.
`permutation`positional-or-keyword	`Sequence[int]`Required	See the operation contract and implementation below.
`optimizer_state`positional-or-keyword	`StateT`Required	See the operation contract and implementation below.
`between_steps`keyword-only	`bool`Required	See the operation contract and implementation below.
`state_adapter`keyword-only	`ExactTransportAdapter[StateT] \| None`Default: None	See the operation contract and implementation below.

Returnstuple[ConfluentPackBank, StateT]

Regression 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.

Full file, imports & fixtures ↗

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 -q

Read the implementation

def permute_bank(
    bank: ConfluentPackBank, permutation: Sequence[int], optimizer_state: StateT,
    *, between_steps: bool, state_adapter: ExactTransportAdapter[StateT] | None = None,
) -> tuple[ConfluentPackBank, StateT]:
    """Move stable IDs, every state row and its moments through one exact permutation."""
    return _exact_transport(bank, optimizer_state, between_steps=between_steps,
                            permutation=tuple(permutation), state_adapter=state_adapter)

FUNCTION

relabel_tanh_sign

omnibias.jax.confluent_bank.relabel_tanh_signSource ↗

relabel_tanh_sign(bank: ConfluentPackBank, slots: Sequence[int], optimizer_state: StateT, *, between_steps: bool, state_adapter: ExactTransportAdapter[StateT] | None=None) -> tuple[ConfluentPackBank, StateT]

Ordinary tanh symmetry, including first-/second-moment transport.

Parameters

Name	Type & default	Contract
`bank`positional-or-keyword	`ConfluentPackBank`Required	See the operation contract and implementation below.
`slots`positional-or-keyword	`Sequence[int]`Required	See the operation contract and implementation below.
`optimizer_state`positional-or-keyword	`StateT`Required	See the operation contract and implementation below.
`between_steps`keyword-only	`bool`Required	See the operation contract and implementation below.
`state_adapter`keyword-only	`ExactTransportAdapter[StateT] \| None`Default: None	See the operation contract and implementation below.

Returnstuple[ConfluentPackBank, StateT]

Regression 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.

Full file, imports & fixtures ↗

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 -q

Read the implementation

def relabel_tanh_sign(
    bank: ConfluentPackBank, slots: Sequence[int], optimizer_state: StateT,
    *, between_steps: bool, state_adapter: ExactTransportAdapter[StateT] | None = None,
) -> tuple[ConfluentPackBank, StateT]:
    """Ordinary tanh symmetry, including first-/second-moment transport."""
    return _exact_transport(bank, optimizer_state, between_steps=between_steps,
                            sign_slots=tuple(slots), state_adapter=state_adapter)

FUNCTION

reset_slot_state

omnibias.jax.confluent_bank.reset_slot_stateSource ↗

reset_slot_state(state: NamedMomentState, bank: ConfluentPackBank, slots: tuple[int, ...]) -> NamedMomentState

Reset affected moment rows, retaining all other rows and the shared clock.

Parameters

Name	Type & default	Contract
`state`positional-or-keyword	`NamedMomentState`Required	See the operation contract and implementation below.
`bank`positional-or-keyword	`ConfluentPackBank`Required	See the operation contract and implementation below.
`slots`positional-or-keyword	`tuple[int, ...]`Required	See the operation contract and implementation below.

ReturnsNamedMomentState

Explicit validation & failure cases

When not isinstance(state, NamedMomentState), raises TypeError('unknown optimizer state requires an explicit OptimizerStateAdapter').
When set(entries) != set(PARAMETER_NAMES), raises ValueError('each named moment must match the complete bank parameter dictionary').
When value.shape != getattr(bank, name).shape, raises ValueError('moment shape does not match parameter').

These are explicit source guards. Backend operations may also raise for invalid shapes, dtypes, or unsupported devices.

Read the implementation

def reset_slot_state(state: NamedMomentState, bank: ConfluentPackBank,
                     slots: tuple[int, ...]) -> NamedMomentState:
    """Reset affected moment rows, retaining all other rows and the shared clock."""
    if not isinstance(state, NamedMomentState):
        raise TypeError("unknown optimizer state requires an explicit OptimizerStateAdapter")
    moments: dict[str, dict[str, Array]] = {}
    for kind, entries in state.moments.items():
        if set(entries) != set(PARAMETER_NAMES):
            raise ValueError("each named moment must match the complete bank parameter dictionary")
        moments[kind] = {}
        for name, value in entries.items():
            if value.shape != getattr(bank, name).shape:
                raise ValueError("moment shape does not match parameter")
            moments[kind][name] = value.at[jnp.asarray(slots)].set(0)
    return NamedMomentState(state.step, moments)

FUNCTION

transport_optimizer_state

omnibias.jax.confluent_bank.transport_optimizer_stateSource ↗

transport_optimizer_state(state: NamedMomentState, bank: ConfluentPackBank, permutation: tuple[int, ...] | None, sign_slots: tuple[int, ...]) -> NamedMomentState

Permute all moment rows; sign-transform named first moments only.

Parameters

Name	Type & default	Contract
`state`positional-or-keyword	`NamedMomentState`Required	See the operation contract and implementation below.
`bank`positional-or-keyword	`ConfluentPackBank`Required	See the operation contract and implementation below.
`permutation`positional-or-keyword	`tuple[int, ...] \| None`Required	See the operation contract and implementation below.
`sign_slots`positional-or-keyword	`tuple[int, ...]`Required	See the operation contract and implementation below.

ReturnsNamedMomentState

Explicit validation & failure cases

When not isinstance(state, NamedMomentState), raises TypeError('unknown optimizer state requires an explicit ExactTransportAdapter').
When set(entries) != set(PARAMETER_NAMES), raises ValueError('each named moment must match the complete bank parameter dictionary').
When sign_slots and kind not in odd | even, raises TypeError('unknown moment parity requires an explicit ExactTransportAdapter').
When value.shape != getattr(bank, name).shape, raises ValueError('moment shape does not match parameter').

These are explicit source guards. Backend operations may also raise for invalid shapes, dtypes, or unsupported devices.

Read the implementation

def transport_optimizer_state(
    state: NamedMomentState, bank: ConfluentPackBank,
    permutation: tuple[int, ...] | None, sign_slots: tuple[int, ...],
) -> NamedMomentState:
    """Permute all moment rows; sign-transform named first moments only."""
    if not isinstance(state, NamedMomentState):
        raise TypeError("unknown optimizer state requires an explicit ExactTransportAdapter")
    odd = {"first", "momentum", "gradient_average"}
    even = {"second", "variance", "max_second", "sum_squares", "absolute"}
    moments: dict[str, dict[str, Array]] = {}
    for kind, entries in state.moments.items():
        if set(entries) != set(PARAMETER_NAMES):
            raise ValueError("each named moment must match the complete bank parameter dictionary")
        if sign_slots and kind not in odd | even:
            raise TypeError("unknown moment parity requires an explicit ExactTransportAdapter")
        moments[kind] = {}
        for name, value in entries.items():
            if value.shape != getattr(bank, name).shape:
                raise ValueError("moment shape does not match parameter")
            result = value if permutation is None else value[jnp.asarray(permutation)]
            if sign_slots and kind in odd and name in ("weights", "scales"):
                result = result.at[jnp.asarray(sign_slots)].multiply(-1)
            moments[kind][name] = result
    return NamedMomentState(state.step, moments)

← Package overview Runnable learning examples →

omnibias.jax.confluent_bank

ConfluentPackBank​

Parameters

Fields & methods

Parameters

Parameters

Parameters

Parameters

Parameters

Parameters

Parameters

Parameters

Parameters

Parameters

Parameters

ExactTransportAdapter​

Fields & methods

Parameters

NamedMomentState​

Parameters

Fields & methods

Parameters

OptimizerStateAdapter​

Fields & methods

Parameters

PARAMETER_NAMES​

STATE_NAMES​

bank_forward​

Parameters

bank_from_checkpoint​

Parameters

bank_slot_terms​

Parameters

commit_transition​

Parameters

detect_transitions​

Parameters

init_confluent_bank​

Parameters

migrate_legacy​

Parameters

permute_bank​

Parameters

relabel_tanh_sign​

Parameters

reset_slot_state​

Parameters

transport_optimizer_state​

Parameters

ConfluentPackBank

ExactTransportAdapter

NamedMomentState

OptimizerStateAdapter

PARAMETER_NAMES

STATE_NAMES

bank_forward

bank_from_checkpoint

bank_slot_terms

commit_transition

detect_transitions

init_confluent_bank

migrate_legacy

permute_bank

relabel_tanh_sign

reset_slot_state

transport_optimizer_state