Decoders¶

Decoder is a torch.nn.Module that realizes the Kleisli coalgebra Vec_D → Kern(T_Σ). Given an input vector, it defines a distribution over terms of a signature through two operations:

sample(vec, ctx, sort) draws a single Term.
log_prob(term, vec, ctx, sort) scores an observed term under the same distribution.

Corecursion over a signature Σ proceeds as follows:

At each sort position, the decoder produces logits over its choice set, every constructor and binder whose codomain is that sort, plus the built-in BoundVar whenever the context contains at least one in-scope variable of that sort.
For the chosen op, the parent vector is split into per-child sub-vectors by the per-(sort, arity) factor function, and the decoder recurses on each child.
Data-sorted children are sampled from a closed vocabulary via the per-sort primitive head; index-sorted children are sampled via binder_select over the in-scope variables.
Binder ops extend Γ before recursing on their scoped arguments, as in the encoder.

Construction sets a recursion-depth bound. At that bound, the decoder restricts the choice set to terminating ops. If a sort has no such op, the decoder raises an error that identifies the sort.

decoder ¶

Decoder runtime: Kleisli coalgebras Vec_D -> Kern(T_Σ).

A Decoder is a torch.nn.Module exposing two operations:

sample: draws a single Term from the distribution induced by an input vector.
log_prob: scores an observed term under the same distribution.

The corecursion structure over a signature Σ is:

At each sort position, the decoder produces logits over its choice set: every constructor and binder whose codomain is that sort, plus the built-in BOUND_VAR_OP whenever the context Γ contains at least one in-scope variable of that sort.
For the chosen op, the parent vector is split into per-child sub-vectors by the per-(sort, arity) factor function, and the decoder recurses on each child under the same canonical form used by the encoder.
Data-sorted children are sampled from a closed vocabulary via the per-sort primitive head; index-sorted children are sampled via binder_select over the in-scope variables.
Binder ops extend Γ before recursing on their scoped arguments, exactly mirroring the encoder.

Termination is depth-bounded at construction. At the budget limit the choice set is restricted to recursion-terminating ops (every child sort is data / index, never object); if no such op exists at a sort, the decoder raises.

No silent type coercion or sentinel value is ever emitted: an observed term whose shape doesn't match the canonical form raises.

Decoder ¶

Decoder(name: str, signature: Signature, sort_dims: dict[str, int], depth: int, structure_fns: dict[str, Callable[[Tensor], Tensor]], primitive_fns: dict[str, Callable[[Tensor], Tensor]], factor_fns: dict[str, dict[int, Callable[[Tensor], tuple[Tensor, ...]]]], binder_select_fn: Callable[[Tensor, list[Tensor]], Tensor], data_vocab: dict[str, list[DataLeaf]], modules_owned: list[Module] | None = None)

Bases: Module

A Kleisli coalgebraic decoder over a signature.

PARAMETER	DESCRIPTION
`name`	Identifier used in diagnostics. TYPE: `str`
`signature`	The Σ whose terms this decoder generates. TYPE: `Signature`
`sort_dims`	Per-sort embedding dimension. TYPE: `dict[str, int]`
`depth`	Hard upper bound on recursion depth. Sampling beyond this depth is restricted to recursion-terminating ops at each sort. TYPE: `int`
`structure_fns`	Per-sort logit producers `(vec) -> Tensor`. Indexed by sort name, plus a shared `""` entry consulted when no sort-specific entry exists. TYPE:* `dict[str, callable]`
`primitive_fns`	Per-data-sort logit producers over the closed token vocab. TYPE: `dict[str, callable]`
`factor_fns`	Per-sort, per-arity child-vector projections `(parent_vec) -> tuple[Tensor, …]` of `n` sub-vectors. TYPE: `dict[str, dict[int, callable]]`
`binder_select_fn`	`(parent_vec, list_of_var_embeddings) -> Tensor` of logits over the in-scope variables of a sort, used by `BOUND_VAR_OP` and by index-sorted child positions. TYPE: `callable`
`data_vocab`	Per-data-sort closed vocabulary aligned with the column order of the corresponding `primitive_fns` output. TYPE: `dict[str, list]`
`modules_owned`	Modules holding the parameters of the logit producers and projections, registered as submodules so their parameters train. TYPE: `list[Module] or None` DEFAULT: `None`

RAISES	DESCRIPTION
`ValueError`	If `depth` is not positive.

Source code in src/quivers/structural/decoder.py

def __init__(
    self,
    name: str,
    signature: Signature,
    sort_dims: dict[str, int],
    depth: int,
    structure_fns: dict[str, Callable[[torch.Tensor], torch.Tensor]],
    primitive_fns: dict[str, Callable[[torch.Tensor], torch.Tensor]],
    factor_fns: dict[
        str, dict[int, Callable[[torch.Tensor], tuple[torch.Tensor, ...]]]
    ],
    binder_select_fn: Callable[[torch.Tensor, list[torch.Tensor]], torch.Tensor],
    data_vocab: dict[str, list[DataLeaf]],
    modules_owned: list[nn.Module] | None = None,
) -> None:
    super().__init__()
    self.name = name
    self.signature = signature
    self.sort_dims = dict(sort_dims)
    if depth <= 0:
        raise ValueError(f"decoder {name!r}: depth must be positive, got {depth}")
    self.depth = depth
    self.structure_fns = dict(structure_fns)
    self.primitive_fns = dict(primitive_fns)
    self.factor_fns = {s: dict(fs) for s, fs in factor_fns.items()}
    self.binder_select_fn = binder_select_fn
    self.data_vocab = dict(data_vocab)
    self._candidates_by_sort = self._collect_candidates()
    for i, m in enumerate(modules_owned or []):
        self.add_module(f"_dec_{i}", m)

sample ¶

sample(vec: Tensor, ctx: Context | None = None, sort: str | None = None) -> Term

Draw one term from the decoder's distribution.

PARAMETER	DESCRIPTION
`vec`	The embedding to decode from. TYPE: `Tensor`
`ctx`	The scope of in-scope bound variables; `EMPTY_CONTEXT` when omitted. TYPE: `Context or None` DEFAULT: `None`
`sort`	The sort of the term to draw; the first declared object sort when omitted. TYPE: `str or None` DEFAULT: `None`

RETURNS	DESCRIPTION
`Term`	The sampled term.

RAISES	DESCRIPTION
`RuntimeError`	If no sort is given and the signature has no object sort, or the decoder lacks a function, vocabulary, or candidate it needs at some position.

Source code in src/quivers/structural/decoder.py

def sample(
    self,
    vec: torch.Tensor,
    ctx: Context | None = None,
    sort: str | None = None,
) -> Term:
    """Draw one term from the decoder's distribution.

    Parameters
    ----------
    vec : torch.Tensor
        The embedding to decode from.
    ctx : Context or None
        The scope of in-scope bound variables; `EMPTY_CONTEXT`
        when omitted.
    sort : str or None
        The sort of the term to draw; the first declared object
        sort when omitted.

    Returns
    -------
    Term
        The sampled term.

    Raises
    ------
    RuntimeError
        If no sort is given and the signature has no object sort,
        or the decoder lacks a function, vocabulary, or candidate
        it needs at some position.
    """
    return self._decode_object(
        vec,
        ctx or EMPTY_CONTEXT,
        self._resolve_sort(sort),
        self.depth,
    )

log_prob ¶

log_prob(term: Term, vec: Tensor, ctx: Context | None = None, sort: str | None = None) -> Tensor

Score an observed term under the decoder's distribution.

PARAMETER	DESCRIPTION
`term`	The observed term. TYPE: `Term`
`vec`	The embedding the term is decoded from. TYPE: `Tensor`
`ctx`	The scope of in-scope bound variables; `EMPTY_CONTEXT` when omitted. TYPE: `Context or None` DEFAULT: `None`
`sort`	The sort of `term`; the first declared object sort when omitted. TYPE: `str or None` DEFAULT: `None`

RETURNS	DESCRIPTION
`Tensor`	The scalar log-probability of `term`.

RAISES	DESCRIPTION
`TypeError`	If `term` is not a `Term`, or an argument does not match the kind of its sort.
`RuntimeError`	If the term's shape is outside the decoder's support, for instance an operation not in the choice set of its sort or a data value outside the closed vocabulary.

Source code in src/quivers/structural/decoder.py

def log_prob(
    self,
    term: Term,
    vec: torch.Tensor,
    ctx: Context | None = None,
    sort: str | None = None,
) -> torch.Tensor:
    """Score an observed term under the decoder's distribution.

    Parameters
    ----------
    term : Term
        The observed term.
    vec : torch.Tensor
        The embedding the term is decoded from.
    ctx : Context or None
        The scope of in-scope bound variables; `EMPTY_CONTEXT`
        when omitted.
    sort : str or None
        The sort of ``term``; the first declared object sort when
        omitted.

    Returns
    -------
    torch.Tensor
        The scalar log-probability of ``term``.

    Raises
    ------
    TypeError
        If ``term`` is not a `Term`, or an argument does not match
        the kind of its sort.
    RuntimeError
        If the term's shape is outside the decoder's support, for
        instance an operation not in the choice set of its sort or
        a data value outside the closed vocabulary.
    """
    if not isinstance(term, Term):
        raise TypeError(
            f"decoder {self.name!r}: log_prob expects a Term, got "
            f"{type(term).__name__}"
        )
    return self._logp_object(
        term,
        vec,
        ctx or EMPTY_CONTEXT,
        self._resolve_sort(sort),
        self.depth,
    )