Encoders

Encoder is a torch.nn.Module that realizes an F-algebra homomorphism T_Σ → Vec_D from terms over a signature to fixed-length vectors. The framework supplies the recursion. The analyst may specify the parametric function for each operation or use the compiler-generated two-layer MLPs, whose input dimensions follow the operation arguments.

For binders, the framework threads a typed de Bruijn context Γ through the recursion. It compresses each binder's annotations in the outer context, creates variable embeddings with var_init_fns, and recurses over the scoped arguments in the extended context.

forward_graph compresses graph signatures. It applies an initial embedder for each vertex kind, a finite number of message-passing rounds with functions indexed by edge and vertex kind, and a readout that reduces the final vertex embeddings to one graph-level vector.

encoder

Encoder runtime: F-algebra homomorphisms T_Σ -> Vec_D.

A Encoder is a torch.nn.Module that, given a closed Term over a signature, returns its vector embedding. The recursion structure is supplied by the framework (algebra homomorphism); per-operation parametric functions are supplied by the user (declared in a encoder { … } DSL block) or by the default scaffolding.

Canonical term form

A term over Σ has exactly one of three argument shapes at each position, determined by the codomain sort's kind:

• object: the argument is a Term whose op is a constructor of Σ, a binder of Σ, or the reserved built-in "BoundVar".
• data: the argument is a raw Python value (string, int, float, …) consumed by the encoder's per-data-sort embedder.
• index: the argument is a non-negative integer denoting a de-Bruijn index into the current scope context Γ.

No Term("Data", …) or other wrappings are accepted at runtime; the canonical form is the only form.

The reserved "BoundVar" op is the built-in de-Bruijn reference: Term("BoundVar", (i,)) at an object position reads the i-th in-scope variable's embedding from Γ.

Binders thread a Context of in-scope variables: each entry records the variable's sort, its embedding, and the type term captured at binding. Compressing a binder constructor:

1. Compresses each of the binder's binds arguments in the OUTER context.
2. Synthesises a fresh variable embedding from each via the encoder's var_init function.
3. Pushes the new entries onto Γ and recurses on the scoped arguments under the extended context.

Graph-shaped signatures are compressed by forward_graph: per-vertex-kind init seeds initial embeddings, finitely many message_passing rounds alternate per-edge-kind message and per-vertex-kind update functions, and a readout reduces the final per-vertex embeddings to one fixed-length vector.

PerOpFn dataclass

PerOpFn(op: str, mode: PerOpMode, args: tuple[str, ...], fn: Callable[..., Tensor], state_var: str | None = None, prefix_var: str | None = None)

A per-operation parametric function.

mode is "plain" (algebra-hom), "recurrent" (left-fold across a list-shaped signature) or "attention" (each step sees a running prefix list). For plain rules, fn is called with one positional argument per child embedding; for recurrent / attention the framework threads state / prefix arguments alongside the children.

Held as a plain dataclass (not a didactic Model) because the fn field is a Python callable that doesn't round-trip through panproto's schema translation.

PARAMETER	DESCRIPTION
op	The constructor or binder name the function interprets. TYPE: str
mode	How the function combines its children. TYPE: PerOpMode
args	The names of the operation's arguments, in positional order. TYPE: tuple[str, ...]
fn	The function mapping child embeddings to the operation's embedding. TYPE: Callable[..., Tensor]
state_var	The name of the threaded state argument in "recurrent" mode. TYPE: str or None DEFAULT: None
prefix_var	The name of the running-prefix argument in "attention" mode. TYPE: str or None DEFAULT: None

Encoder

Encoder(name: str, signature: Signature, sort_dims: dict[str, int], op_fns: dict[str, PerOpFn], var_init_fns: dict[tuple[str, str] | str, Callable[[Tensor | None], Tensor]], data_embedders: dict[str, Callable[[DataLeaf], Tensor]], modules_owned: list[Module] | None = None, iterations: int = 0, init_fns: dict[str, Callable[[DataLeaf], Tensor]] | None = None, message_fns: dict[str, Callable[..., Tensor]] | None = None, update_fns: dict[str, Callable[..., Tensor]] | None = None, readout: Callable[[list[Tensor]], Tensor] | None = None)

Bases: Module

An F-algebra homomorphism T_Σ -> Vec_D realised as an nn.Module.

The graph-specific parameters (iterations, init_fns, message_fns, update_fns, readout) are used only by forward_graph on a graph-shaped signature.

PARAMETER	DESCRIPTION
name	Identifier used in diagnostics. TYPE: str
signature	The Σ whose terms this encoder consumes. TYPE: Signature
sort_dims	Per-sort embedding dimension. TYPE: dict[str, int]
op_fns	One entry per constructor and binder of Σ. The DSL compiler scaffolds defaults for omitted ops. TYPE: dict[str, PerOpFn]
var_init_fns	Variable initialisers called at every binder to mint a fresh embedding for each introduced variable. An annotated variable is keyed by (var_sort, annot_sort) and receives the annotation's embedding; an unannotated one is keyed by var_sort and receives None. TYPE: dict[tuple[str, str] | str, Callable]
data_embedders	One entry per data sort: (raw_value) -> embedding. TYPE: dict[str, Callable[[DataLeaf], Tensor]]
modules_owned	Modules holding the parameters of the per-op functions, registered as submodules so their parameters train. TYPE: list[Module] or None DEFAULT: None
iterations	Number of message-passing rounds in forward_graph. TYPE: int DEFAULT: 0
init_fns	Per-vertex-kind initial embedders. TYPE: dict[str, Callable[[DataLeaf], Tensor]] or None DEFAULT: None
message_fns	Per-edge-kind (src_e, tgt_e) -> message functions. TYPE: dict[str, Callable[..., Tensor]] or None DEFAULT: None
update_fns	Per-vertex-kind (self_e, aggregated_msg) -> next_e. TYPE: dict[str, Callable[..., Tensor]] or None DEFAULT: None
readout	(list[Vec_D]) -> Vec_D reducer. TYPE: Callable[[list[Tensor]], Tensor] or None DEFAULT: None

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

def __init__(
    self,
    name: str,
    signature: Signature,
    sort_dims: dict[str, int],
    op_fns: dict[str, PerOpFn],
    var_init_fns: dict[
        tuple[str, str] | str,
        Callable[[torch.Tensor | None], torch.Tensor],
    ],
    data_embedders: dict[str, Callable[[DataLeaf], torch.Tensor]],
    modules_owned: list[nn.Module] | None = None,
    iterations: int = 0,
    init_fns: dict[str, Callable[[DataLeaf], torch.Tensor]] | None = None,
    message_fns: dict[str, Callable[..., torch.Tensor]] | None = None,
    update_fns: dict[str, Callable[..., torch.Tensor]] | None = None,
    readout: Callable[[list[torch.Tensor]], torch.Tensor] | None = None,
) -> None:
    super().__init__()
    self.name = name
    self.signature = signature
    self.sort_dims = dict(sort_dims)
    self.op_fns = dict(op_fns)
    # Keyed by `(var_sort, annot_sort)` for annotated binders or
    # by `var_sort` (a plain string) for unannotated ones.
    self.var_init_fns: dict[
        tuple[str, str] | str,
        Callable[[torch.Tensor | None], torch.Tensor],
    ] = dict(var_init_fns)
    self.data_embedders = dict(data_embedders)
    self.iterations = iterations
    self.init_fns = dict(init_fns or {})
    self.message_fns = dict(message_fns or {})
    self.update_fns = dict(update_fns or {})
    self.readout = readout
    for i, m in enumerate(modules_owned or []):
        self.add_module(f"_op_{i}", m)

forward

forward(term: Term, ctx: Context | None = None) -> Tensor

Encode a term to its embedding by structural recursion.

PARAMETER	DESCRIPTION
term	The term to encode. TYPE: Term
ctx	The scope the term's free bound variables refer to; EMPTY_CONTEXT when omitted. TYPE: Context or None DEFAULT: None

RETURNS	DESCRIPTION
Tensor	The embedding 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 uses an operation, sort, or bound-variable index the encoder cannot interpret.

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

def forward(self, term: Term, ctx: Context | None = None) -> torch.Tensor:
    """Encode a term to its embedding by structural recursion.

    Parameters
    ----------
    term : Term
        The term to encode.
    ctx : Context or None
        The scope the term's free bound variables refer to;
        `EMPTY_CONTEXT` when omitted.

    Returns
    -------
    torch.Tensor
        The embedding 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 uses an operation, sort, or bound-variable
        index the encoder cannot interpret.
    """
    if not isinstance(term, Term):
        raise TypeError(
            f"encoder {self.name!r}: expected a Term at the root, got "
            f"{type(term).__name__} ({term!r}); construct one with "
            f"`make_term` or `bound_var`"
        )
    return self._compress_object(term, ctx or EMPTY_CONTEXT)

forward_graph

forward_graph(vertices: list[tuple[str, DataLeaf]], edges: list[tuple[str, int, int]]) -> Tensor

Encode a graph by message passing followed by a readout.

Each vertex is initialised by its kind's initial embedder; each of iterations rounds sends a message along every edge (both ways for an undirected edge kind), averages each vertex's inbox, and applies its kind's update function.

PARAMETER	DESCRIPTION
vertices	Each vertex's kind and payload. TYPE: list[tuple[str, DataLeaf]]
edges	Each edge's kind, source index, and target index. TYPE: list[tuple[str, int, int]]

RETURNS	DESCRIPTION
Tensor	The readout of the final vertex embeddings.

RAISES	DESCRIPTION
RuntimeError	If the signature is not graph-shaped, iterations is not positive, an edge references an out-of-range vertex or an unknown edge kind, or a needed initial, message, update, or readout function is missing.

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

def forward_graph(
    self,
    vertices: list[tuple[str, DataLeaf]],
    edges: list[tuple[str, int, int]],
) -> torch.Tensor:
    """Encode a graph by message passing followed by a readout.

    Each vertex is initialised by its kind's initial embedder;
    each of ``iterations`` rounds sends a message along every edge
    (both ways for an undirected edge kind), averages each
    vertex's inbox, and applies its kind's update function.

    Parameters
    ----------
    vertices : list[tuple[str, DataLeaf]]
        Each vertex's kind and payload.
    edges : list[tuple[str, int, int]]
        Each edge's kind, source index, and target index.

    Returns
    -------
    torch.Tensor
        The readout of the final vertex embeddings.

    Raises
    ------
    RuntimeError
        If the signature is not graph-shaped, ``iterations`` is not
        positive, an edge references an out-of-range vertex or an
        unknown edge kind, or a needed initial, message, update, or
        readout function is missing.
    """
    if not self.signature.is_graph():
        raise RuntimeError(
            f"encoder {self.name!r}: forward_graph requires a graph signature"
        )
    if self.iterations <= 0:
        raise RuntimeError(
            f"encoder {self.name!r}: graph encoder requires `iterations` > 0"
        )

    embeds: list[torch.Tensor] = []
    for vkind, payload in vertices:
        init = self.init_fns.get(vkind)
        if init is None:
            raise RuntimeError(
                f"encoder {self.name!r}: no init for vertex_kind {vkind!r}"
            )
        embeds.append(init(payload))

    for _ in range(self.iterations):
        inboxes: list[list[torch.Tensor]] = [[] for _ in vertices]
        for ekind, src, tgt in edges:
            if src < 0 or tgt < 0 or src >= len(vertices) or tgt >= len(vertices):
                raise RuntimeError(
                    f"encoder {self.name!r}: edge {(ekind, src, tgt)} "
                    f"references out-of-range vertex"
                )
            edge_spec = self.signature.edge_kinds.get(ekind)
            if edge_spec is None:
                raise RuntimeError(
                    f"encoder {self.name!r}: unknown edge_kind {ekind!r}"
                )
            m_fn = self.message_fns.get(ekind)
            if m_fn is None:
                raise RuntimeError(
                    f"encoder {self.name!r}: no message fn for edge_kind {ekind!r}"
                )
            inboxes[tgt].append(m_fn(embeds[src], embeds[tgt]))
            if not edge_spec.directed:
                inboxes[src].append(m_fn(embeds[tgt], embeds[src]))

        new_embeds = list(embeds)
        for i, (vkind, _) in enumerate(vertices):
            upd = self.update_fns.get(vkind)
            if upd is None:
                raise RuntimeError(
                    f"encoder {self.name!r}: no update fn for vertex_kind {vkind!r}"
                )
            msgs = inboxes[i]
            if msgs:
                agg = torch.stack(msgs, dim=0).mean(dim=0)
            else:
                agg = torch.zeros_like(embeds[i])
            new_embeds[i] = upd(embeds[i], agg)
        embeds = new_embeds

    if self.readout is None:
        raise RuntimeError(
            f"encoder {self.name!r}: graph encoder requires a readout function"
        )
    return self.readout(embeds)

make_default_op_fn

make_default_op_fn(op: str, arg_dims: tuple[int, ...], out_dim: int) -> tuple[Module, Callable[..., Tensor]]

Build a parametric per-op function as a 2-layer MLP over concatenated child embeddings.

arg_dims is the tuple of per-argument dimensions in the order in which the children are passed to the function. For binders this is (annot_dim_1, …, annot_dim_k, scoped_dim_1, …, scoped_dim_m). A nullary operation gets a learned constant vector instead of an MLP.

PARAMETER	DESCRIPTION
op	The operation name, used in parameter names and diagnostics. TYPE: str
arg_dims	The embedding dimension of each child, in call order. TYPE: tuple[int, ...]
out_dim	The dimension of the operation's output embedding. TYPE: int

RETURNS	DESCRIPTION
tuple[Module, Callable[..., Tensor]]	The module owning the parameters, and the function that maps the child embeddings to the output embedding.

RAISES	DESCRIPTION
RuntimeError	Raised by the returned function when it is called with a number of children other than len(arg_dims).

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

def make_default_op_fn(
    op: str,
    arg_dims: tuple[int, ...],
    out_dim: int,
) -> tuple[nn.Module, Callable[..., torch.Tensor]]:
    """Build a parametric per-op function as a 2-layer MLP over
    concatenated child embeddings.

    ``arg_dims`` is the tuple of per-argument dimensions in the
    order in which the children are passed to the function. For
    binders this is ``(annot_dim_1, …, annot_dim_k, scoped_dim_1,
    …, scoped_dim_m)``. A nullary operation gets a learned constant
    vector instead of an MLP.

    Parameters
    ----------
    op : str
        The operation name, used in parameter names and diagnostics.
    arg_dims : tuple[int, ...]
        The embedding dimension of each child, in call order.
    out_dim : int
        The dimension of the operation's output embedding.

    Returns
    -------
    tuple[nn.Module, Callable[..., torch.Tensor]]
        The module owning the parameters, and the function that maps
        the child embeddings to the output embedding.

    Raises
    ------
    RuntimeError
        Raised by the returned function when it is called with a
        number of children other than ``len(arg_dims)``.
    """
    arity = len(arg_dims)
    if arity == 0:
        param = nn.Parameter(torch.randn(out_dim) * 0.1)
        mod = nn.Module()
        mod.register_parameter(f"const_{op}", param)

        def call() -> torch.Tensor:
            return param

        return mod, call

    in_dim = sum(arg_dims)
    hidden = max(out_dim, 64)
    mlp = nn.Sequential(
        nn.Linear(in_dim, hidden),
        nn.Tanh(),
        nn.Linear(hidden, out_dim),
    )

    def call(*children: torch.Tensor) -> torch.Tensor:
        if len(children) != arity:
            raise RuntimeError(
                f"op {op!r}: expected {arity} children, got {len(children)}"
            )
        cat = torch.cat([c.reshape(-1) for c in children], dim=-1)
        return mlp(cat)

    return mlp, call

make_default_var_init

make_default_var_init(in_dim: int, out_dim: int) -> tuple[Module, Callable[[Tensor], Tensor]]

A parametric var_init(ty) function: an MLP mapping a type embedding to a fresh variable embedding.

PARAMETER	DESCRIPTION
in_dim	The dimension of the type-annotation embedding. TYPE: int
out_dim	The dimension of the variable embedding. TYPE: int

RETURNS	DESCRIPTION
tuple[Module, Callable[[Tensor], Tensor]]	The MLP owning the parameters, and the function that maps a type embedding to a variable embedding.

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

def make_default_var_init(
    in_dim: int, out_dim: int
) -> tuple[nn.Module, Callable[[torch.Tensor], torch.Tensor]]:
    """A parametric `var_init(ty)` function: an MLP mapping a type
    embedding to a fresh variable embedding.

    Parameters
    ----------
    in_dim : int
        The dimension of the type-annotation embedding.
    out_dim : int
        The dimension of the variable embedding.

    Returns
    -------
    tuple[nn.Module, Callable[[torch.Tensor], torch.Tensor]]
        The MLP owning the parameters, and the function that maps a
        type embedding to a variable embedding.
    """
    mlp = nn.Sequential(
        nn.Linear(in_dim, max(out_dim, 64)),
        nn.Tanh(),
        nn.Linear(max(out_dim, 64), out_dim),
    )

    def call(ty: torch.Tensor) -> torch.Tensor:
        return mlp(ty.reshape(-1))

    return mlp, call

Type aliases and constants

Structural compression: signatures, encoders, decoders, losses.

Runtime substrate for QVR signatures, encoders, decoders, and attached losses: a uniform algebraic interface for compressing structured objects to fixed-length vectors and decoding them under a learned distribution.

BOUND_VAR_OP module-attribute

BOUND_VAR_OP = 'BoundVar'

The reserved op name of a bound-variable term.

A Term whose op is this name refers to a variable a binder introduced. The signature compiler forbids it as a user-declared constructor or binder name.

PerOpMode

PerOpMode = Literal['plain', 'recurrent', 'attention']

How a PerOpFn combines its children: "plain" applies the function once, "recurrent" left-folds it across a list-shaped signature, and "attention" gives each step the running prefix.