omnibias.keras.blocks.operator
Keras 3 operator layers and activation paths implemented with keras.ops for TensorFlow, JAX, or Torch.
import omnibias.keras.blocks.operatorModule 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.
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.OpNameomnibias.keras.blocks.OpNameomnibias.keras.blocks.operator.OpName
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
| Name | Type & default | Contract |
|---|---|---|
oppositional-or-keyword | OpNameDefault: 'identity' | one of identity / grad / laplacian / derivative / band / integral |
basepositional-or-keyword | str | ActivationSpec[Any]Default: 'gaussian' | str or ActivationSpec, default "gaussian" |
channelspositional-or-keyword | intDefault: 1 | int, default 1 |
init_biaspositional-or-keyword | floatDefault: 0.0 | See the operation contract and implementation below. |
init_deltapositional-or-keyword | float | NoneDefault: None | 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_biasespositional-or-keyword | boolDefault: True | bool, default True |
learnable_signspositional-or-keyword | bool | NoneDefault: None | bool, optional |
derivative_orderpositional-or-keyword | int | NoneDefault: None | int, optional Required for op="derivative" (block uses K = order + 1). |
normalize_integralpositional-or-keyword | boolDefault: False | bool, default False |
integral_small_widthpositional-or-keyword | floatDefault: 0.0001 | float, default 1e-4 |
kwargsvariadic keyword | AnyRequired | See 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" -qRegression 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 -qFields & 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) -> NoneSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
oppositional-or-keyword | OpNameDefault: 'identity' | See the operation contract and implementation below. |
basepositional-or-keyword | str | ActivationSpec[Any]Default: 'gaussian' | See the operation contract and implementation below. |
channelspositional-or-keyword | intDefault: 1 | See the operation contract and implementation below. |
init_biaspositional-or-keyword | floatDefault: 0.0 | See the operation contract and implementation below. |
init_deltapositional-or-keyword | float | NoneDefault: None | See the operation contract and implementation below. |
learnable_biasespositional-or-keyword | boolDefault: True | See the operation contract and implementation below. |
learnable_signspositional-or-keyword | bool | NoneDefault: None | See the operation contract and implementation below. |
derivative_orderpositional-or-keyword | int | NoneDefault: None | See the operation contract and implementation below. |
normalize_integralpositional-or-keyword | boolDefault: False | See the operation contract and implementation below. |
integral_small_widthpositional-or-keyword | floatDefault: 0.0001 | See the operation contract and implementation below. |
kwargsvariadic keyword | AnyRequired | See the operation contract and implementation below. |
Returns: None
Validation & failure cases
- When
op not in _VALID_OPS, raisesValueError(f'op must be one of {list(_VALID_OPS)}, got {op!r}.'). - When
derivative_order is not None and op in ('grad', 'laplacian'), raisesValueError(f"derivative_order is only valid with op='derivative'; use op={op!r} without it."). - When
derivative_order is None, raisesValueError("OperatorBlock(op='derivative') requires derivative_order."). - When
derivative_order < 1, raisesValueError(f'derivative_order must be >= 1, got {derivative_order}.'). - When
spec.fastpath is None, raisesTypeError(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, raisesTypeError(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" -qMethod 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 = Truechannelsproperty
channels(self) -> intSee the class contract above and the exact method implementation below.
Returns: int
Method implementation
def channels(self) -> int:
return self.ombu.num_channelsKproperty
K(self) -> intSee 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" -qMethod implementation
def K(self) -> int:
return self.ombu.Kbase_nameproperty
base_name(self) -> strSee the class contract above and the exact method implementation below.
Returns: str
Method implementation
def base_name(self) -> str:
return self.ombu.spec.namederivative_orderproperty
derivative_order(self) -> intSee the class contract above and the exact method implementation below.
Returns: int
Method implementation
def derivative_order(self) -> int:
return self._derivative_ordercallmethod
call(self, z: Any) -> AnyDispatch to the appropriate forward path for this op type.
Parameters
| Name | Type & default | Contract |
|---|---|---|
zpositional-or-keyword | AnyRequired | See 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] == 5python -m pytest "packages/omnibias-keras/tests/test_blocks.py::test_cmbconv1d_shape" -qMethod 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)literal_forwardmethod
literal_forward(self, z: Any) -> AnyLiteral activation-space counterpart for diagnostics.
Parameters
| Name | Type & default | Contract |
|---|---|---|
zpositional-or-keyword | AnyRequired | See 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)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 configopattribute
op: OpName = opone of identity / grad / laplacian / derivative / band / integral
normalize_integralattribute
normalize_integral: bool = normalize_integralbool, default False
View source ↗integral_small_widthattribute
integral_small_width: float = integral_small_widthfloat, default 1e-4
View source ↗init_deltaattribute
init_delta: float | None = init_deltaInitial 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.
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.
builtattribute
built = TruePublic instance attribute assigned by __init__. The source assignment is built = True; it describes instance state, not a literal constructor default.
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 configOther public import paths
omnibias.keras.OperatorBlockomnibias.keras.blocks.OperatorBlockomnibias.keras.blocks.operator.OperatorBlock