Skip to main content

omnibias.keras.blocks.operator

0.0.2a1AlphaPython >=3.10Apache-2.0

Keras 3 operator layers and activation paths implemented with keras.ops for TensorFlow, JAX, or Torch.

import omnibias.keras.blocks.operator
Module contract & mathematical conventions

Operator-typed wrapper around the Keras OperatorMultiBiasUnit.

An OperatorBlock carries an operator role tag -- "identity", "grad", "laplacian", "derivative", "band", or "integral" -- that selects both the multi-bias arity K and the forward behaviour. Mirrors omnibias.torch.blocks.operator.OperatorBlock.

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.

CONSTANT

OpName​

omnibias.keras.OpNameSource ↗
OpName = Literal['identity', 'grad', 'laplacian', 'derivative', 'band', 'integral']

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
OpName = Literal["identity", "grad", "laplacian", "derivative", "band", "integral"]
Other public import paths
  • omnibias.keras.OpName
  • omnibias.keras.blocks.OpName
  • omnibias.keras.blocks.operator.OpName
CLASS

OperatorBlock​

omnibias.keras.OperatorBlockSource ↗
class OperatorBlock(layers.Layer)

An OMBU with operator-type semantics (Keras backend).

Derivative-op semantics (important precondition)

The derivative ops (grad, laplacian, derivative) evaluate the closed-form **mean-bias fast-path** sigma^(order)(z + mean(biases)) (see omnibias.keras.unit.OperatorMultiBiasUnit.analytic_derivative), *not* the literal finite-difference stencil sum_k s_k sigma(z + b_k) over the individual biases. The two coincide only under the **identity-nesting** initialisation -- init_delta = 0 (the block's default for these ops), which seeds all K bias columns equal so the mean *is* the shared bias and the closed-form sigma^(order) at that point is the intended derivative.

With **untied biases** that have drifted apart during training (or a nonzero init_delta), mean(biases) no longer characterises the stencil and the op returns the single-point closed-form derivative *at the mean bias* rather than the literal multi-bias stencil. Keep init_delta = 0 (and, if exact stencil semantics matter, learnable_biases=False) for the derivative ops; the activation-level math stays mathematically equivalent to the torch / jax backends under this precondition.

Parameters

op : one of identity / grad / laplacian / derivative / band / integral base : str or ActivationSpec, default "gaussian" channels : int, default 1 init_delta : float, optional Initial bias spread. Defaults to 0 for derivative ops (the identity-nesting precondition above) and 1 for band / integral. A nonzero value breaks the mean-bias == literal-stencil equivalence for the derivative ops. learnable_biases : bool, default True learnable_signs : bool, optional derivative_order : int, optional Required for op="derivative" (block uses K = order + 1). normalize_integral : bool, default False integral_small_width : float, default 1e-4

Backend scope: this identifies the corresponding mathematical construction. It does not guarantee bitwise-equal floating-point results across devices, compilers, dtypes, or backend versions. Validate the selected path with an appropriate absolute and relative tolerance.

Parameters

NameType & defaultContract
oppositional-or-keywordOpNameDefault: 'identity'one of identity / grad / laplacian / derivative / band / integral
basepositional-or-keywordstr | ActivationSpec[Any]Default: 'gaussian'str or ActivationSpec, default "gaussian"
channelspositional-or-keywordintDefault: 1int, default 1
init_biaspositional-or-keywordfloatDefault: 0.0See the operation contract and implementation below.
init_deltapositional-or-keywordfloat | NoneDefault: Nonefloat, optional Initial bias spread. Defaults to 0 for derivative ops (the identity-nesting precondition above) and 1 for band / integral. A nonzero value breaks the mean-bias == literal-stencil equivalence for the derivative ops.
learnable_biasespositional-or-keywordboolDefault: Truebool, default True
learnable_signspositional-or-keywordbool | NoneDefault: Nonebool, optional
derivative_orderpositional-or-keywordint | NoneDefault: Noneint, optional Required for op="derivative" (block uses K = order + 1).
normalize_integralpositional-or-keywordboolDefault: Falsebool, default False
integral_small_widthpositional-or-keywordfloatDefault: 0.0001float, default 1e-4
kwargsvariadic keywordAnyRequiredSee the operation contract and implementation below.
Executed regression · invalid op raises

Recorded coverage confirms this test executed the API body. 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_invalid_op_raises() -> None:
    with pytest.raises(ValueError):
        OperatorBlock(channels=2, op="nonsense")
python -m pytest "packages/omnibias-keras/tests/test_blocks.py::test_invalid_op_raises" -q
Regression example · integral block runs

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_integral_block_runs() -> None:
    z = np.random.default_rng(4).normal(size=(8, 4))
    block = OperatorBlock(channels=4, op="integral", base="gaussian", init_delta=1.0)
    out = _np(block(_t(z)))
    assert out.shape == (8, 4)
    assert np.all(np.isfinite(out))
python -m pytest packages/omnibias-keras/tests/test_blocks.py -k test_integral_block_runs -q

Fields & methods

__init__method
__init__(self, op: OpName='identity', base: str | ActivationSpec[Any]='gaussian', channels: int=1, init_bias: float=0.0, init_delta: float | None=None, learnable_biases: bool=True, learnable_signs: bool | None=None, derivative_order: int | None=None, normalize_integral: bool=False, integral_small_width: float=0.0001, **kwargs: Any) -> None

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

Parameters

NameType & defaultContract
oppositional-or-keywordOpNameDefault: 'identity'See the operation contract and implementation below.
basepositional-or-keywordstr | ActivationSpec[Any]Default: 'gaussian'See the operation contract and implementation below.
channelspositional-or-keywordintDefault: 1See the operation contract and implementation below.
init_biaspositional-or-keywordfloatDefault: 0.0See the operation contract and implementation below.
init_deltapositional-or-keywordfloat | NoneDefault: NoneSee the operation contract and implementation below.
learnable_biasespositional-or-keywordboolDefault: TrueSee the operation contract and implementation below.
learnable_signspositional-or-keywordbool | NoneDefault: NoneSee the operation contract and implementation below.
derivative_orderpositional-or-keywordint | NoneDefault: NoneSee the operation contract and implementation below.
normalize_integralpositional-or-keywordboolDefault: FalseSee the operation contract and implementation below.
integral_small_widthpositional-or-keywordfloatDefault: 0.0001See the operation contract and implementation below.
kwargsvariadic keywordAnyRequiredSee the operation contract and implementation below.

Returns: None

Validation & failure cases
  • When op not in _VALID_OPS, raises ValueError(f'op must be one of {list(_VALID_OPS)}, got {op!r}.').
  • When derivative_order is not None and op in ('grad', 'laplacian'), raises ValueError(f"derivative_order is only valid with op='derivative'; use op={op!r} without it.").
  • When derivative_order is None, raises ValueError("OperatorBlock(op='derivative') requires derivative_order.").
  • When derivative_order < 1, raises ValueError(f'derivative_order must be >= 1, got {derivative_order}.').
  • When spec.fastpath is None, raises TypeError(f'OperatorBlock(op={op!r}) requires a base activation with a closed-form derivative kernel; activation {spec.name!r} has none.').
  • When op == 'integral' and spec.integral is None, raises TypeError(f"OperatorBlock(op='integral') requires a base activation with a closed-form integral kernel; activation {spec.name!r} has none.").
Executed regression · invalid op raises

Recorded coverage confirms this test executed the API body. 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_invalid_op_raises() -> None:
    with pytest.raises(ValueError):
        OperatorBlock(channels=2, op="nonsense")
python -m pytest "packages/omnibias-keras/tests/test_blocks.py::test_invalid_op_raises" -q
Method implementation
    def __init__(
        self,
        op: OpName = "identity",
        base: str | ActivationSpec[Any] = "gaussian",
        channels: int = 1,
        init_bias: float = 0.0,
        init_delta: float | None = None,
        learnable_biases: bool = True,
        learnable_signs: bool | None = None,
        derivative_order: int | None = None,
        normalize_integral: bool = False,
        integral_small_width: float = 1e-4,
        **kwargs: Any,
    ) -> None:
        super().__init__(**kwargs)
        if op not in _VALID_OPS:
            raise ValueError(f"op must be one of {list(_VALID_OPS)}, got {op!r}.")
        self.op = op
        self.normalize_integral = normalize_integral
        self.integral_small_width = integral_small_width
        if derivative_order is not None and op in ("grad", "laplacian"):
            raise ValueError(
                f"derivative_order is only valid with op='derivative'; use op={op!r} without it."
            )
        if op == "derivative":
            if derivative_order is None:
                raise ValueError("OperatorBlock(op='derivative') requires derivative_order.")
            if derivative_order < 1:
                raise ValueError(f"derivative_order must be >= 1, got {derivative_order}.")
            K = derivative_order + 1
            self._derivative_order = derivative_order
        else:
            K = _OP_K[op]
            self._derivative_order = _OP_DERIVATIVE_ORDER[op]
        spec = base if isinstance(base, ActivationSpec) else get_activation(base)
        self._base_name = base if isinstance(base, str) else base.name

        if op in ("grad", "laplacian", "derivative"):
            required_order = self._derivative_order
            if spec.fastpath is None:
                raise TypeError(
                    f"OperatorBlock(op={op!r}) requires a base activation with a "
                    f"closed-form derivative kernel; activation {spec.name!r} has none."
                )
            try:
                spec.fastpath(ops.zeros((1,)), required_order)
            except NotImplementedError as e:
                raise TypeError(
                    f"OperatorBlock(op={op!r}) requires order-{required_order} "
                    f"derivative; activation {spec.name!r} only supports a partial "
                    f"fastpath. ({e})"
                ) from None
        if op == "integral" and spec.integral is None:
            raise TypeError(
                f"OperatorBlock(op='integral') requires a base activation with a "
                f"closed-form integral kernel; activation {spec.name!r} has none."
            )

        if init_delta is None:
            init_delta = 1.0 if op in ("band", "integral") else 0.0
        if learnable_signs is None:
            learnable_signs = False
        self.init_delta = init_delta
        self.learnable_biases = learnable_biases
        self.learnable_signs = learnable_signs

        dtype = self.variable_dtype
        if op == "band":
            signs_init = _window_signs(channels, K, dtype=dtype)
            offsets = central_bias_offsets(K, init_delta, dtype=dtype)[None, :]
            biases_init = identity_init_biases(channels, K, bias_value=init_bias, dtype=dtype) + offsets
        elif op == "integral":
            signs_init = _window_signs(channels, K, dtype=dtype)
            biases_init = np.empty((channels, K), dtype=dtype)
            biases_init[:, 0] = init_bias
            biases_init[:, 1] = _inverse_softplus(init_delta)
        else:
            signs_init = None
            biases_init = None

        self.ombu = OperatorMultiBiasUnit(
            num_channels=channels,
            K=K,
            base=spec,
            init_bias=init_bias,
            init_delta=init_delta if op not in ("band", "integral") else 0.0,
            learnable_biases=learnable_biases,
            learnable_signs=learnable_signs,
            biases_init=biases_init,
            signs_init=signs_init,
        )
        self.built = True
View source ↗
channelsproperty
channels(self) -> int

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

Returns: int

Method implementation
    def channels(self) -> int:
        return self.ombu.num_channels
View source ↗
Kproperty
K(self) -> int

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

Returns: int

Executed regression · derivative block arbitrary order

Recorded coverage confirms this test executed the API body. 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_derivative_block_arbitrary_order() -> None:
    z = np.random.default_rng(3).normal(size=(8, 4))
    block = OperatorBlock(channels=4, op="derivative", base="tanh", derivative_order=3)
    assert block.K == 4
    np.testing.assert_allclose(
        _np(block(_t(z))),
        _np(get_activation("tanh").fastpath(_t(z), 3)),
        rtol=1e-9,
        atol=1e-10,
    )
python -m pytest "packages/omnibias-keras/tests/test_blocks.py::test_derivative_block_arbitrary_order" -q
Method implementation
    def K(self) -> int:
        return self.ombu.K
View source ↗
base_nameproperty
base_name(self) -> str

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

Returns: str

Method implementation
    def base_name(self) -> str:
        return self.ombu.spec.name
View source ↗
derivative_orderproperty
derivative_order(self) -> int

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

Returns: int

Method implementation
    def derivative_order(self) -> int:
        return self._derivative_order
View source ↗
callmethod
call(self, z: Any) -> Any

Dispatch to the appropriate forward path for this op type.

Parameters

NameType & defaultContract
zpositional-or-keywordAnyRequiredSee the operation contract and implementation below.

Returns: Any

Executed regression · cmbconv1d shape

Recorded coverage confirms this test executed the API body. 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_cmbconv1d_shape() -> None:
    x = np.random.default_rng(6).normal(size=(4, 10, 3))
    layer = cmbConv1D(filters=5, kernel_size=3, op="identity", base="tanh")
    out = _np(layer(_t(x)))
    assert out.shape[0] == 4 and out.shape[-1] == 5
python -m pytest "packages/omnibias-keras/tests/test_blocks.py::test_cmbconv1d_shape" -q
Method implementation
    def call(self, z: Any) -> Any:
        """Dispatch to the appropriate forward path for this op type."""
        if self.op in ("identity", "band"):
            return self.ombu(z)
        if self.op == "integral":
            return self.ombu.analytic_integral(
                z,
                strict_window=True,
                normalize=self.normalize_integral,
                small_width_threshold=self.integral_small_width,
            )
        return self.ombu.analytic_derivative(z, order=self._derivative_order)
View source ↗
literal_forwardmethod
literal_forward(self, z: Any) -> Any

Literal activation-space counterpart for diagnostics.

Parameters

NameType & defaultContract
zpositional-or-keywordAnyRequiredSee the operation contract and implementation below.

Returns: Any

Method implementation
    def literal_forward(self, z: Any) -> Any:
        """Literal activation-space counterpart for diagnostics."""
        if self.op == "integral":
            center = self.ombu.biases[:, 0]
            width = ops.softplus(self.ombu.biases[:, 1])
            half_width = 0.5 * width
            endpoints = ops.stack((center - half_width, center + half_width), axis=-1)
            return multibias_literal_forward(
                z, endpoints, self.ombu.signs, self.ombu.spec.forward
            )
        return self.ombu(z)
View source ↗
get_configmethod
get_config(self) -> dict[str, Any]

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

Returns: dict[str, Any]

Method implementation
    def get_config(self) -> dict[str, Any]:
        config = super().get_config()
        config.update(
            {
                "op": self.op,
                "base": self._base_name,
                "channels": self.channels,
                "init_delta": self.init_delta,
                "learnable_biases": self.learnable_biases,
                "learnable_signs": self.learnable_signs,
                "derivative_order": (
                    self._derivative_order if self.op == "derivative" else None
                ),
                "normalize_integral": self.normalize_integral,
                "integral_small_width": self.integral_small_width,
            }
        )
        return config
View source ↗
opattribute
op: OpName = op

one of identity / grad / laplacian / derivative / band / integral

View source ↗
normalize_integralattribute
normalize_integral: bool = normalize_integral

bool, default False

View source ↗
integral_small_widthattribute
integral_small_width: float = integral_small_width

float, default 1e-4

View source ↗
init_deltaattribute
init_delta: float | None = init_delta

Initial bias spread. Defaults to 0 for derivative ops (the identity-nesting precondition above) and 1 for band / integral. A nonzero value breaks the mean-bias == literal-stencil equivalence for the derivative ops.

View source ↗
learnable_biasesattribute
learnable_biases: bool = learnable_biases

bool, default True

View source ↗
learnable_signsattribute
learnable_signs: bool | None = learnable_signs

bool, optional

View source ↗
ombuattribute
ombu = OperatorMultiBiasUnit(num_channels=channels, K=K, base=spec, init_bias=init_bias, init_delta=init_delta if op not in ('band', 'integral') else 0.0, learnable_biases=learnable_biases, learnable_signs=learnable_signs, biases_init=biases_init, signs_init=signs_init)

Public instance attribute assigned by __init__. The source assignment is ombu = OperatorMultiBiasUnit(num_channels=channels, K=K, base=spec, init_bias=init_bias, init_delta=init_delta if op not in ('band', 'integral') else 0.0, learnable_biases=learnable_biases, learnable_signs=learnable_signs, biases_init=biases_init, signs_init=signs_init); it describes instance state, not a literal constructor default.

View source ↗
builtattribute
built = True

Public instance attribute assigned by __init__. The source assignment is built = True; it describes instance state, not a literal constructor default.

View source ↗
Read the implementation
class OperatorBlock(layers.Layer):
    """An OMBU with operator-type semantics (Keras backend).

    Derivative-op semantics (important precondition)
    ------------------------------------------------
    The derivative ops (``grad``, ``laplacian``, ``derivative``) evaluate the
    closed-form **mean-bias fast-path** ``sigma^(order)(z + mean(biases))`` (see
    :meth:`omnibias.keras.unit.OperatorMultiBiasUnit.analytic_derivative`), *not*
    the literal finite-difference stencil ``sum_k s_k sigma(z + b_k)`` over the
    individual biases. The two coincide only under the **identity-nesting**
    initialisation -- ``init_delta = 0`` (the block's default for these ops),
    which seeds all ``K`` bias columns equal so the mean *is* the shared bias and
    the closed-form ``sigma^(order)`` at that point is the intended derivative.

    With **untied biases** that have drifted apart during training (or a nonzero
    ``init_delta``), ``mean(biases)`` no longer characterises the stencil and the
    op returns the single-point closed-form derivative *at the mean bias* rather
    than the literal multi-bias stencil. Keep ``init_delta = 0`` (and, if exact
    stencil semantics matter, ``learnable_biases=False``) for the derivative ops;
    the activation-level math stays bit-identical to the torch / jax backends
    under this precondition.

    Parameters
    ----------
    op : one of ``identity / grad / laplacian / derivative / band / integral``
    base : str or :class:`ActivationSpec`, default ``"gaussian"``
    channels : int, default 1
    init_delta : float, optional
        Initial bias spread. Defaults to ``0`` for derivative ops (the
        identity-nesting precondition above) and ``1`` for ``band`` / ``integral``.
        A nonzero value breaks the mean-bias == literal-stencil equivalence for
        the derivative ops.
    learnable_biases : bool, default True
    learnable_signs : bool, optional
    derivative_order : int, optional
        Required for ``op="derivative"`` (block uses ``K = order + 1``).
    normalize_integral : bool, default False
    integral_small_width : float, default 1e-4
    """

    def __init__(
        self,
        op: OpName = "identity",
        base: str | ActivationSpec[Any] = "gaussian",
        channels: int = 1,
        init_bias: float = 0.0,
        init_delta: float | None = None,
        learnable_biases: bool = True,
        learnable_signs: bool | None = None,
        derivative_order: int | None = None,
        normalize_integral: bool = False,
        integral_small_width: float = 1e-4,
        **kwargs: Any,
    ) -> None:
        super().__init__(**kwargs)
        if op not in _VALID_OPS:
            raise ValueError(f"op must be one of {list(_VALID_OPS)}, got {op!r}.")
        self.op = op
        self.normalize_integral = normalize_integral
        self.integral_small_width = integral_small_width
        if derivative_order is not None and op in ("grad", "laplacian"):
            raise ValueError(
                f"derivative_order is only valid with op='derivative'; use op={op!r} without it."
            )
        if op == "derivative":
            if derivative_order is None:
                raise ValueError("OperatorBlock(op='derivative') requires derivative_order.")
            if derivative_order < 1:
                raise ValueError(f"derivative_order must be >= 1, got {derivative_order}.")
            K = derivative_order + 1
            self._derivative_order = derivative_order
        else:
            K = _OP_K[op]
            self._derivative_order = _OP_DERIVATIVE_ORDER[op]
        spec = base if isinstance(base, ActivationSpec) else get_activation(base)
        self._base_name = base if isinstance(base, str) else base.name

        if op in ("grad", "laplacian", "derivative"):
            required_order = self._derivative_order
            if spec.fastpath is None:
                raise TypeError(
                    f"OperatorBlock(op={op!r}) requires a base activation with a "
                    f"closed-form derivative kernel; activation {spec.name!r} has none."
                )
            try:
                spec.fastpath(ops.zeros((1,)), required_order)
            except NotImplementedError as e:
                raise TypeError(
                    f"OperatorBlock(op={op!r}) requires order-{required_order} "
                    f"derivative; activation {spec.name!r} only supports a partial "
                    f"fastpath. ({e})"
                ) from None
        if op == "integral" and spec.integral is None:
            raise TypeError(
                f"OperatorBlock(op='integral') requires a base activation with a "
                f"closed-form integral kernel; activation {spec.name!r} has none."
            )

        if init_delta is None:
            init_delta = 1.0 if op in ("band", "integral") else 0.0
        if learnable_signs is None:
            learnable_signs = False
        self.init_delta = init_delta
        self.learnable_biases = learnable_biases
        self.learnable_signs = learnable_signs

        dtype = self.variable_dtype
        if op == "band":
            signs_init = _window_signs(channels, K, dtype=dtype)
            offsets = central_bias_offsets(K, init_delta, dtype=dtype)[None, :]
            biases_init = identity_init_biases(channels, K, bias_value=init_bias, dtype=dtype) + offsets
        elif op == "integral":
            signs_init = _window_signs(channels, K, dtype=dtype)
            biases_init = np.empty((channels, K), dtype=dtype)
            biases_init[:, 0] = init_bias
            biases_init[:, 1] = _inverse_softplus(init_delta)
        else:
            signs_init = None
            biases_init = None

        self.ombu = OperatorMultiBiasUnit(
            num_channels=channels,
            K=K,
            base=spec,
            init_bias=init_bias,
            init_delta=init_delta if op not in ("band", "integral") else 0.0,
            learnable_biases=learnable_biases,
            learnable_signs=learnable_signs,
            biases_init=biases_init,
            signs_init=signs_init,
        )
        self.built = True

    @property
    def channels(self) -> int:
        return self.ombu.num_channels

    @property
    def K(self) -> int:
        return self.ombu.K

    @property
    def base_name(self) -> str:
        return self.ombu.spec.name

    @property
    def derivative_order(self) -> int:
        return self._derivative_order

    def call(self, z: Any) -> Any:
        """Dispatch to the appropriate forward path for this op type."""
        if self.op in ("identity", "band"):
            return self.ombu(z)
        if self.op == "integral":
            return self.ombu.analytic_integral(
                z,
                strict_window=True,
                normalize=self.normalize_integral,
                small_width_threshold=self.integral_small_width,
            )
        return self.ombu.analytic_derivative(z, order=self._derivative_order)

    def literal_forward(self, z: Any) -> Any:
        """Literal activation-space counterpart for diagnostics."""
        if self.op == "integral":
            center = self.ombu.biases[:, 0]
            width = ops.softplus(self.ombu.biases[:, 1])
            half_width = 0.5 * width
            endpoints = ops.stack((center - half_width, center + half_width), axis=-1)
            return multibias_literal_forward(
                z, endpoints, self.ombu.signs, self.ombu.spec.forward
            )
        return self.ombu(z)

    def get_config(self) -> dict[str, Any]:
        config = super().get_config()
        config.update(
            {
                "op": self.op,
                "base": self._base_name,
                "channels": self.channels,
                "init_delta": self.init_delta,
                "learnable_biases": self.learnable_biases,
                "learnable_signs": self.learnable_signs,
                "derivative_order": (
                    self._derivative_order if self.op == "derivative" else None
                ),
                "normalize_integral": self.normalize_integral,
                "integral_small_width": self.integral_small_width,
            }
        )
        return config
Other public import paths
  • omnibias.keras.OperatorBlock
  • omnibias.keras.blocks.OperatorBlock
  • omnibias.keras.blocks.operator.OperatorBlock