Handlers

Conditioning, intervention, reweighting, blocking, replay, and lifting.

clamp

Clamp handler: condition named sample sites on given values.

ClampHandler is the effect-stack analogue of Pyro's condition. A site whose name appears in its data is answered with the given value and scored under its own distribution, so the value contributes the site's density to the joint and the trace records the site as observed. The name avoids collision with the top-level quivers.inference.conditioning.condition factory, which returns a Conditioned model wrapper.

ClampHandler

ClampHandler(data: dict[str, Tensor])

Bases: EffectHandler

Condition named sample sites on given values.

PARAMETER	DESCRIPTION
data	Site name to the value the site is clamped to. Other sites pass through untouched. TYPE: dict[str, Tensor]

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

def __init__(self, data: dict[str, torch.Tensor]) -> None:
    self.data = dict(data)

install

install(run: RunContext) -> tuple[Installation, ...]

Install a conditioning handler of the random instance.

PARAMETER	DESCRIPTION
run	The run being prepared. TYPE: RunContext

RETURNS	DESCRIPTION
tuple[Installation, ...]	The conditioning handler, which forwards sites it has no value for and ignores values at sites the program never samples.

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

def install(self, run: RunContext) -> tuple[Installation, ...]:
    """Install a conditioning handler of the ``random`` instance.

    Parameters
    ----------
    run : RunContext
        The run being prepared.

    Returns
    -------
    tuple[Installation, ...]
        The conditioning handler, which forwards sites it has no
        value for and ignores values at sites the program never
        samples.
    """
    kernel = run.kernel
    return (
        Installation(
            kernel.random,
            condition_handler(
                self.data,
                score_instance=kernel.score.entry.instance,
                result_validator=run.validator,
                missing=MissingValuePolicy.FORWARD,
                extra=ExtraValuePolicy.IGNORE,
                answer_type=run.result_type,
                key=f"clamp-{id(self):x}",
            ),
        ),
    )

clamp

clamp(data: dict[str, Tensor]) -> ClampHandler

Return a ClampHandler conditioning on the given site values.

PARAMETER	DESCRIPTION
data	Site name to value. TYPE: dict[str, Tensor]

RETURNS	DESCRIPTION
ClampHandler	The handler.

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

def clamp(data: dict[str, torch.Tensor]) -> ClampHandler:
    """Return a `ClampHandler` conditioning on the given site values.

    Parameters
    ----------
    data : dict[str, torch.Tensor]
        Site name to value.

    Returns
    -------
    ClampHandler
        The handler.
    """
    return ClampHandler(data)

do

Intervention handler: fix named sites without scoring them.

DoHandler implements Pearl's do operator on the effect stack. A site whose name appears in its data is answered with the given value and contributes nothing to the joint: under intervention its distribution is replaced by a point mass. Downstream sites see the intervened value, and a trace records the site as deterministic.

DoHandler

DoHandler(data: dict[str, Tensor])

Bases: EffectHandler

Intervene on named sites.

PARAMETER	DESCRIPTION
data	Site name to the value the site is fixed to. TYPE: dict[str, Tensor]

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

def __init__(self, data: dict[str, torch.Tensor]) -> None:
    self.data = dict(data)

install

install(run: RunContext) -> tuple[Installation, ...]

Install an intervening handler of the random instance.

PARAMETER	DESCRIPTION
run	The run being prepared; each intervened site's address is recorded on it, so a trace marks the site deterministic. TYPE: RunContext

RETURNS	DESCRIPTION
tuple[Installation, ...]	The intervening handler.

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

def install(self, run: RunContext) -> tuple[Installation, ...]:
    """Install an intervening handler of the ``random`` instance.

    Parameters
    ----------
    run : RunContext
        The run being prepared; each intervened site's address is
        recorded on it, so a trace marks the site deterministic.

    Returns
    -------
    tuple[Installation, ...]
        The intervening handler.
    """

    def mark(request: RuntimeRequest, value: object) -> None:
        """Record an intervened site on the run.

        Parameters
        ----------
        request : RuntimeRequest
            The site's request.
        value : object
            The value fixed, unused.
        """
        del value
        run.interventions.add(run.key_of(request))

    return (
        Installation(
            run.kernel.random,
            intervene_handler(
                self.data,
                result_validator=run.validator,
                on_intervene=mark,
                answer_type=run.result_type,
                key=f"do-{id(self):x}",
            ),
        ),
    )

do

do(data: dict[str, Tensor]) -> DoHandler

Return a DoHandler intervening on the given site values.

PARAMETER	DESCRIPTION
data	Site name to value. TYPE: dict[str, Tensor]

RETURNS	DESCRIPTION
DoHandler	The handler.

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

def do(data: dict[str, torch.Tensor]) -> DoHandler:
    """Return a `DoHandler` intervening on the given site values.

    Parameters
    ----------
    data : dict[str, torch.Tensor]
        Site name to value.

    Returns
    -------
    DoHandler
        The handler.
    """
    return DoHandler(data)

scale

Scale handler: multiply every density contribution by a factor.

ScaleHandler is a transformer of the program's score instance: every contribution any site makes passes through it on the way to the run's accumulator, multiplied by the factor. Stacking scale with mask composes the two elementwise products.

ScaleHandler

ScaleHandler(factor: float)

Bases: EffectHandler

Multiply every density contribution by a scalar.

PARAMETER	DESCRIPTION
factor	The multiplier. TYPE: float

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

def __init__(self, factor: float) -> None:
    self.factor = float(factor)

install

install(run: RunContext) -> tuple[Installation, ...]

Install a scaling transformer of the score instance.

PARAMETER	DESCRIPTION
run	The run being prepared. TYPE: RunContext

RETURNS	DESCRIPTION
tuple[Installation, ...]	The transformer.

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

def install(self, run: RunContext) -> tuple[Installation, ...]:
    """Install a scaling transformer of the ``score`` instance.

    Parameters
    ----------
    run : RunContext
        The run being prepared.

    Returns
    -------
    tuple[Installation, ...]
        The transformer.
    """
    factor = self.factor

    def scaled(weight: object) -> object:
        """Scale one contribution.

        Parameters
        ----------
        weight : object
            The contribution.

        Returns
        -------
        object
            The contribution times the factor.
        """
        return torch.as_tensor(weight) * factor

    return (
        Installation(
            run.kernel.score,
            reweight_handler(
                scaled,
                score_instance=run.kernel.score.entry.instance,
                answer_type=run.result_type,
                key=f"scale-{id(self):x}",
            ),
        ),
    )

scale

scale(factor: float) -> ScaleHandler

Return a ScaleHandler with the given factor.

PARAMETER	DESCRIPTION
factor	The multiplier. TYPE: float

RETURNS	DESCRIPTION
ScaleHandler	The handler.

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

def scale(factor: float) -> ScaleHandler:
    """Return a `ScaleHandler` with the given factor.

    Parameters
    ----------
    factor : float
        The multiplier.

    Returns
    -------
    ScaleHandler
        The handler.
    """
    return ScaleHandler(factor)

mask

Mask handler: gate every density contribution elementwise.

MaskHandler is a transformer of the program's score instance: the mask is broadcast against every contribution on its way to the run's accumulator. A boolean mask switches entries; a float mask scales them.

MaskHandler

MaskHandler(mask_tensor: Tensor)

Bases: EffectHandler

Gate every density contribution elementwise.

PARAMETER	DESCRIPTION
mask_tensor	The mask, broadcast against each contribution. TYPE: Tensor

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

def __init__(self, mask_tensor: torch.Tensor) -> None:
    self.mask = mask_tensor

install

install(run: RunContext) -> tuple[Installation, ...]

Install a masking transformer of the score instance.

PARAMETER	DESCRIPTION
run	The run being prepared. TYPE: RunContext

RETURNS	DESCRIPTION
tuple[Installation, ...]	The transformer.

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

def install(self, run: RunContext) -> tuple[Installation, ...]:
    """Install a masking transformer of the ``score`` instance.

    Parameters
    ----------
    run : RunContext
        The run being prepared.

    Returns
    -------
    tuple[Installation, ...]
        The transformer.
    """
    mask = self.mask

    def masked(weight: object) -> object:
        """Mask one contribution.

        Parameters
        ----------
        weight : object
            The contribution.

        Returns
        -------
        object
            The contribution times the mask.
        """
        return torch.as_tensor(weight) * mask.to(torch.as_tensor(weight).dtype)

    return (
        Installation(
            run.kernel.score,
            reweight_handler(
                masked,
                score_instance=run.kernel.score.entry.instance,
                answer_type=run.result_type,
                key=f"mask-{id(self):x}",
            ),
        ),
    )

mask

mask(mask_tensor: Tensor) -> MaskHandler

Return a MaskHandler with the given mask.

PARAMETER	DESCRIPTION
mask_tensor	The mask. TYPE: Tensor

RETURNS	DESCRIPTION
MaskHandler	The handler.

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

def mask(mask_tensor: torch.Tensor) -> MaskHandler:
    """Return a `MaskHandler` with the given mask.

    Parameters
    ----------
    mask_tensor : torch.Tensor
        The mask.

    Returns
    -------
    MaskHandler
        The handler.
    """
    return MaskHandler(mask_tensor)

block

Block handler: hide sites from the handlers outside it.

BlockHandler answers hidden sites itself, drawing and scoring them exactly as the run's default handler would, so no handler outside it sees their requests: an outer trace does not record them and an outer clamp cannot condition them. Every other site forwards.

BlockHandler

BlockHandler(hide: list[str] | None = None, expose: list[str] | None = None)

Bases: EffectHandler

Hide sites from outer handlers.

PARAMETER	DESCRIPTION
hide	Site names to hide; every site when both arguments are absent. TYPE: list[str] or None DEFAULT: None
expose	Site names to leave visible, hiding every other. TYPE: list[str] or None DEFAULT: None

RAISES	DESCRIPTION
ValueError	If both hide and expose are given.

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

def __init__(
    self, hide: list[str] | None = None, expose: list[str] | None = None
) -> None:
    if hide is not None and expose is not None:
        raise ValueError(
            "BlockHandler: pass at most one of `hide` and `expose`, not both."
        )
    self.hide = list(hide) if hide is not None else None
    self.expose = list(expose) if expose is not None else None

install

install(run: RunContext) -> tuple[Installation, ...]

Install a blocking handler of the random instance.

PARAMETER	DESCRIPTION
run	The run being prepared. TYPE: RunContext

RETURNS	DESCRIPTION
tuple[Installation, ...]	The blocking handler.

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

def install(self, run: RunContext) -> tuple[Installation, ...]:
    """Install a blocking handler of the ``random`` instance.

    Parameters
    ----------
    run : RunContext
        The run being prepared.

    Returns
    -------
    tuple[Installation, ...]
        The blocking handler.
    """
    kernel = run.kernel
    return (
        Installation(
            kernel.random,
            block_handler(
                self._should_block,
                score_instance=kernel.score.entry.instance,
                result_validator=run.validator,
                answer_type=run.result_type,
                key=f"block-{id(self):x}",
            ),
        ),
    )

block

block(hide: list[str] | None = None, expose: list[str] | None = None) -> BlockHandler

Return a BlockHandler hiding or exposing the named sites.

PARAMETER	DESCRIPTION
hide	Site names to hide. TYPE: list[str] or None DEFAULT: None
expose	Site names to leave visible. TYPE: list[str] or None DEFAULT: None

RETURNS	DESCRIPTION
BlockHandler	The handler.

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

def block(
    hide: list[str] | None = None,
    expose: list[str] | None = None,
) -> BlockHandler:
    """Return a `BlockHandler` hiding or exposing the named sites.

    Parameters
    ----------
    hide : list[str] or None
        Site names to hide.
    expose : list[str] or None
        Site names to leave visible.

    Returns
    -------
    BlockHandler
        The handler.
    """
    return BlockHandler(hide=hide, expose=expose)

replay

Replay handler: reinstall the values a trace recorded.

ReplayHandler answers each site the trace holds with the recorded value and scores it under the site's current distribution, so replay fixes the values and not the densities: the joint of the replayed run is the model's density at the recorded values.

ReplayHandler

ReplayHandler(trace: Trace)

Bases: EffectHandler

Replay the values of a recorded trace.

PARAMETER	DESCRIPTION
trace	The trace whose site values are reinstalled. TYPE: Trace

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

def __init__(self, trace: Trace) -> None:
    self._trace = trace

install

install(run: RunContext) -> tuple[Installation, ...]

Install a scoring replay handler of the random instance.

PARAMETER	DESCRIPTION
run	The run being prepared. TYPE: RunContext

RETURNS	DESCRIPTION
tuple[Installation, ...]	The replay handler, which forwards sites the trace lacks and ignores recorded sites the program never samples.

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

def install(self, run: RunContext) -> tuple[Installation, ...]:
    """Install a scoring replay handler of the ``random`` instance.

    Parameters
    ----------
    run : RunContext
        The run being prepared.

    Returns
    -------
    tuple[Installation, ...]
        The replay handler, which forwards sites the trace lacks and
        ignores recorded sites the program never samples.
    """
    kernel = run.kernel
    values = {
        name: site.value
        for name, site in self._trace.sites.items()
        if not site.is_deterministic or site.morphism is not None
    }
    return (
        Installation(
            kernel.random,
            replay_handler(
                values,
                result_validator=run.validator,
                policy=ReplayPolicy.CLAMP_AND_SCORE,
                score_instance=kernel.score.entry.instance,
                extra=ExtraValuePolicy.IGNORE,
                answer_type=run.result_type,
                key=f"replay-{id(self):x}",
            ),
        ),
    )

replay

replay(trace: Trace) -> ReplayHandler

Return a ReplayHandler replaying the given trace.

PARAMETER	DESCRIPTION
trace	The trace. TYPE: Trace

RETURNS	DESCRIPTION
ReplayHandler	The handler.

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

def replay(trace: Trace) -> ReplayHandler:
    """Return a `ReplayHandler` replaying the given trace.

    Parameters
    ----------
    trace : Trace
        The trace.

    Returns
    -------
    ReplayHandler
        The handler.
    """
    return ReplayHandler(trace)

lift

Lift handler: draw every learned parameter from a prior.

LiftHandler handles the program's param instance: each parameter a morphism reads through Param.get is answered with a draw from a Normal(value, prior_scale) prior centred on its current value, one draw per parameter per run, and the morphism runs under the drawn values. The draws are kept in sampled_params keyed by the site and the parameter's qualified name.

LiftHandler

LiftHandler(prior_scale: float = 1.0)

Bases: EffectHandler

Draw every reachable parameter from a Normal prior.

PARAMETER	DESCRIPTION
prior_scale	Standard deviation of the Normal prior around each parameter's current value. TYPE: float DEFAULT: 1.0

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

def __init__(self, prior_scale: float = 1.0) -> None:
    self.prior_scale = float(prior_scale)
    self.sampled_params: dict[str, torch.Tensor] = {}

install

install(run: RunContext) -> tuple[Installation, ...]

Install a prior-drawing handler of the param instance.

PARAMETER	DESCRIPTION
run	The run being prepared, whose parameter store holds each parameter's current value. TYPE: RunContext

RETURNS	DESCRIPTION
tuple[Installation, ...]	The handler, which answers each parameter's first read with a fresh draw and later reads with the same draw.

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

def install(self, run: RunContext) -> tuple[Installation, ...]:
    """Install a prior-drawing handler of the ``param`` instance.

    Parameters
    ----------
    run : RunContext
        The run being prepared, whose parameter store holds each
        parameter's current value.

    Returns
    -------
    tuple[Installation, ...]
        The handler, which answers each parameter's first read with
        a fresh draw and later reads with the same draw.
    """
    scale = self.prior_scale
    sampled = self.sampled_params

    def draw(name: str, request: RuntimeRequest) -> object:
        """Answer a parameter read with its prior draw.

        Parameters
        ----------
        name : str
            The parameter's qualified name.
        request : RuntimeRequest
            The read, unused.

        Returns
        -------
        object
            The draw made for the name on this run.
        """
        del request
        if name not in sampled:
            current = run.parameters[name]
            sampled[name] = current.detach() + scale * torch.randn_like(current)
        return sampled[name]

    return (
        Installation(
            run.kernel.param,
            param_handler(
                draw,
                result_validator=run.validator,
                total=False,
                answer_type=run.result_type,
                key=f"lift-{id(self):x}",
            ),
        ),
    )

reset

reset() -> None

Clear the draws so the next run makes fresh ones.

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

def reset(self) -> None:
    """Clear the draws so the next run makes fresh ones."""
    self.sampled_params.clear()

lift

lift(prior_scale: float = 1.0) -> LiftHandler

Return a LiftHandler with the given prior scale.

PARAMETER	DESCRIPTION
prior_scale	Standard deviation of the prior. TYPE: float DEFAULT: 1.0

RETURNS	DESCRIPTION
LiftHandler	The handler.

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

def lift(prior_scale: float = 1.0) -> LiftHandler:
    """Return a `LiftHandler` with the given prior scale.

    Parameters
    ----------
    prior_scale : float
        Standard deviation of the prior.

    Returns
    -------
    LiftHandler
        The handler.
    """
    return LiftHandler(prior_scale)