Checked programs

A program morphism that runs its checked computation on the reference machine.

checked_program

A program morphism that runs its checked computation.

A program that calls a named computation has no step table the runtime compiler can build: the call's body is a term of the checked module, and only the reference machine runs it. Its compiled form is a CheckedProgram, a morphism whose forward samples and joint densities are the module's own sample_program and run_program runs, so the program's Python surface is the one entry point the command line and the REPL invoke.

CheckedProgram

CheckedProgram(domain: AnySpace, codomain: AnySpace, module: QiecModule, name: str)

Bases: ContinuousMorphism

A program run as its elaborated computation on the reference machine.

PARAMETER	DESCRIPTION
domain	The program's declared domain. TYPE: AnySpace
codomain	The program's declared codomain. TYPE: AnySpace
module	The checked module holding the program's entry point. TYPE: QiecModule
name	The program's name. TYPE: str

Source code in src/quivers/effects/checked_program.py

def __init__(
    self, domain: AnySpace, codomain: AnySpace, module: QiecModule, name: str
) -> None:
    super().__init__(domain, codomain)
    self._module = module
    self._name = name
    self._entry: ProgramEntry = program_entry(module, name)

module property

module: QiecModule

The checked module the program runs in.

name property

name: str

The program's name.

entry property

entry: ProgramEntry

The program's elaborated entry point.

observed_names

observed_names() -> set[str]

The names the program observes.

RETURNS	DESCRIPTION
set[str]	The labels of its observe sites.

Source code in src/quivers/effects/checked_program.py

def observed_names(self) -> set[str]:
    """The names the program observes.

    Returns
    -------
    set[str]
        The labels of its ``observe`` sites.
    """
    return {site.name for site in self._entry.sites if site.kind == "observe"}

rsample

rsample(x: Tensor, sample_shape: Size = Size(), observations: Mapping[str, Tensor] | None = None) -> Tensor

Run the program forward, once per input row and sample.

PARAMETER	DESCRIPTION
x	Program input. Shape (batch, ...). TYPE: Tensor
sample_shape	Additional leading sample dimensions; each element is one independent run. TYPE: Size DEFAULT: Size()
observations	The program's data by parameter name, and values for the sample sites the runs condition on; every other site is drawn. TYPE: Mapping[str, Tensor] or None DEFAULT: None

RETURNS	DESCRIPTION
Tensor	The returned values, shape (*sample_shape, batch, ...).

Source code in src/quivers/effects/checked_program.py

def rsample(  # type: ignore[override]
    self,
    x: torch.Tensor,
    sample_shape: torch.Size = torch.Size(),
    observations: Mapping[str, torch.Tensor] | None = None,
) -> torch.Tensor:
    """Run the program forward, once per input row and sample.

    Parameters
    ----------
    x : torch.Tensor
        Program input. Shape ``(batch, ...)``.
    sample_shape : torch.Size
        Additional leading sample dimensions; each element is one
        independent run.
    observations : Mapping[str, torch.Tensor] or None
        The program's data by parameter name, and values for the
        sample sites the runs condition on; every other site is
        drawn.

    Returns
    -------
    torch.Tensor
        The returned values, shape ``(*sample_shape, batch, ...)``.
    """
    given = dict(observations or {})
    batch = int(x.shape[0]) if x.dim() >= 1 else 1
    runs = math.prod(sample_shape)
    rows: list[torch.Tensor] = []
    for _ in range(max(runs, 1)):
        values: list[torch.Tensor] = []
        for row in range(batch):
            data, sites = self._split(x, row, given)
            run = sample_program(self._module, self._name, data=data, sites=sites)
            values.append(_tensor(run.value))
        rows.append(torch.stack(values))
    stacked = torch.stack(rows)
    if not sample_shape:
        return stacked[0]
    return stacked.reshape((*sample_shape, *stacked.shape[1:]))

has_conditional_density

has_conditional_density() -> bool

No: a program's density at its output marginalizes its draws.

RETURNS	DESCRIPTION
bool	Always False; log_joint scores the draws themselves.

Source code in src/quivers/effects/checked_program.py

def has_conditional_density(self) -> bool:
    """No: a program's density at its output marginalizes its draws.

    Returns
    -------
    bool
        Always ``False``;
        [`log_joint`][quivers.effects.checked_program.CheckedProgram.log_joint]
        scores the draws themselves.
    """
    return False

log_prob

log_prob(x: Tensor, y: Tensor) -> Tensor

Log-probability is not defined for a program's output alone.

PARAMETER	DESCRIPTION
x	Program input. TYPE: Tensor
y	Program output. TYPE: Tensor

RAISES	DESCRIPTION
NotImplementedError	Always: the density at the output marginalizes every draw.

Source code in src/quivers/effects/checked_program.py

def log_prob(self, x: torch.Tensor, y: torch.Tensor) -> torch.Tensor:
    """Log-probability is not defined for a program's output alone.

    Parameters
    ----------
    x : torch.Tensor
        Program input.
    y : torch.Tensor
        Program output.

    Raises
    ------
    NotImplementedError
        Always: the density at the output marginalizes every draw.
    """
    del x, y
    raise NotImplementedError(
        "log_prob is not supported for programs; computing p(y | x) "
        "requires marginalizing over all intermediate draws. Use "
        "log_joint with a value for every site."
    )

log_joint

log_joint(x: Tensor, intermediates: Mapping[str, Tensor]) -> Tensor

The joint log-density at given values for every site.

PARAMETER	DESCRIPTION
x	Program input. Shape (batch, ...). TYPE: Tensor
intermediates	The program's data by parameter name and a value for every sample site, by label. TYPE: Mapping[str, Tensor]

RETURNS	DESCRIPTION
Tensor	The log joint of each input row's run. Shape (batch,). The machine scores numbers, so the result carries no gradient path to the given tensors.

RAISES	DESCRIPTION
ExecutionFailure	If a site the program reaches is given no value, or a given site is never reached.

Source code in src/quivers/effects/checked_program.py

def log_joint(
    self, x: torch.Tensor, intermediates: Mapping[str, torch.Tensor]
) -> torch.Tensor:
    """The joint log-density at given values for every site.

    Parameters
    ----------
    x : torch.Tensor
        Program input. Shape ``(batch, ...)``.
    intermediates : Mapping[str, torch.Tensor]
        The program's data by parameter name and a value for every
        sample site, by label.

    Returns
    -------
    torch.Tensor
        The log joint of each input row's run. Shape ``(batch,)``.
        The machine scores numbers, so the result carries no
        gradient path to the given tensors.

    Raises
    ------
    ExecutionFailure
        If a site the program reaches is given no value, or a given
        site is never reached.
    """
    batch = int(x.shape[0]) if x.dim() >= 1 else 1
    totals: list[torch.Tensor] = []
    for row in range(batch):
        data, sites = self._split(x, row, intermediates)
        run = run_program(self._module, self._name, data=data, sites=sites)
        totals.append(
            torch.as_tensor(run.log_joint, dtype=torch.get_default_dtype())
        )
    return torch.stack(totals)