StoppingRules

steerability.algorithms.output_control.stopping_rules

args

control

StoppingRules

Bases: OutputControl

Stop rules as configuration: substring, token, and budget stops.

StoppingRules is the smallest member of the generic family. It is sampling-expressible: the pipeline merges its configuration into the call's normalized generation parameters (export_generation_params), and the backend session composes the resulting stop rules, so the control runs on every backend:

- `stop_texts=["\n\nQ:"]` halts a row once its continuation contains the substring.
- `stop_token_ids=[13]` halts a row once its last generated token is one of the ids.
- `budget=64` tightens `max_new_tokens` to at most `budget`.

Token ids are returned as generated (the stop text plus any token-boundary overrun stays in the ids); the pipeline truncates decoded text at the first stop-string occurrence and rows halted by these rules report finish_reason="stop" (budget stops report "length"). get_stopping_criteria remains available for direct composition outside the pipeline and returns fresh criteria anchored at the current prompt length.

Source code in steerability/algorithms/output_control/stopping_rules/control.py

class StoppingRules(OutputControl):
    """Stop rules as configuration: substring, token, and budget stops.

    `StoppingRules` is the smallest member of the generic family. It is sampling-expressible:
    the pipeline merges its configuration into the call's normalized generation parameters
    (`export_generation_params`), and the backend session composes the resulting stop rules,
    so the control runs on every backend:

        - `stop_texts=["\\n\\nQ:"]` halts a row once its continuation contains the substring.
        - `stop_token_ids=[13]` halts a row once its last generated token is one of the ids.
        - `budget=64` tightens `max_new_tokens` to at most `budget`.

    Token ids are returned as generated (the stop text plus any token-boundary overrun stays in
    the ids); the pipeline truncates decoded text at the first stop-string occurrence and rows
    halted by these rules report `finish_reason="stop"` (budget stops report `"length"`).
    `get_stopping_criteria` remains available for direct composition outside the pipeline and
    returns fresh criteria anchored at the current prompt length.
    """

    Args = StoppingRulesArgs

    supports_batching: bool = True

    tokenizer: PreTrainedTokenizerBase | None = None

    def steer(self, model: PreTrainedModel, tokenizer: PreTrainedTokenizerBase | None = None, **_) -> PreTrainedModel:
        """Attach the tokenizer (required whenever `stop_texts` is configured)."""
        self.tokenizer = tokenizer or getattr(model, "tokenizer", None)
        if self.stop_texts and self.tokenizer is None:
            raise RuntimeError("StoppingRules requires a tokenizer when 'stop_texts' is configured.")
        return model

    def requirements(self) -> Requirements:
        """Stop rules are session contract on every backend, so no phase requires anything."""
        return Requirements()

    def export_generation_params(self, runtime_kwargs: dict | None = None) -> Mapping[str, Any]:
        """The configured stops as normalized generation parameters."""
        contribution: dict[str, Any] = {}
        if self.stop_texts:
            contribution["stop_strings"] = tuple(self.stop_texts)
        if self.stop_token_ids:
            contribution["stop_token_ids"] = tuple(self.stop_token_ids)
        if self.budget is not None:
            contribution["max_new_tokens"] = self.budget
        return contribution

    def get_stopping_criteria(self, input_ids, runtime_kwargs, **kwargs) -> list:
        """Return fresh criteria anchored at the current prompt length."""
        prompt_len = input_ids.size(1)
        criteria = []
        for text in self.stop_texts:
            criteria.append(StopOnSubstring(self.tokenizer, text, prompt_len))
        if self.stop_token_ids:
            criteria.append(StopOnTokens(self.stop_token_ids))
        if self.budget is not None:
            criteria.append(BudgetTokens(self.budget, prompt_len))
        return criteria

RUNTIME_KWARGS_SCHEMA = [] class-attribute instance-attribute

args = self.Args.validate(*args, **kwargs) instance-attribute

enabled = True class-attribute instance-attribute

include_in_scoring = True class-attribute instance-attribute

same_model_forwards = False class-attribute instance-attribute

supports_batching = True class-attribute instance-attribute

tokenizer = None class-attribute instance-attribute

cleanup()

Release resources allocated during steer().

Override in subclasses that allocate GPU memory or other resources during steering. Default no-op.

Source code in steerability/algorithms/core/base_control.py

def cleanup(self) -> None:
    """Release resources allocated during `steer()`.

    Override in subclasses that allocate GPU memory or other resources during steering. Default
    no-op.
    """
    pass

clone_for_call(seed=None)

A configuration-preserving shallow clone for one generation call.

The clone shares steer-time artifacts (memories, steering vectors, attached tokenizers) with the original but has its own attribute namespace, so per-call attribute mutation on the clone never races another call using the original. When seed is given and the control defines reseed(seed), the clone's client-side RNG is re-seeded.

Parameters:

Name	Type	Description	Default
seed	int | None	Optional seed forwarded to the clone's reseed().	None

Returns:

Type	Description
BaseControl	The clone.

Source code in steerability/algorithms/core/base_control.py

def clone_for_call(self, seed: int | None = None) -> "BaseControl":
    """A configuration-preserving shallow clone for one generation call.

    The clone shares steer-time artifacts (memories, steering vectors, attached tokenizers)
    with the original but has its own attribute namespace, so per-call attribute mutation on
    the clone never races another call using the original. When `seed` is given and the
    control defines `reseed(seed)`, the clone's client-side RNG is re-seeded.

    Args:
        seed: Optional seed forwarded to the clone's `reseed()`.

    Returns:
        The clone.
    """
    clone = copy.copy(self)
    if seed is not None:
        reseed = getattr(clone, "reseed", None)
        if callable(reseed):
            reseed(seed)
    return clone

export_constraint(runtime_kwargs=None)

The control's declarative constrained-decoding source, or None.

A control whose per-step masking compiles from a declarative source returns a ConstraintSource; on a backend advertising Capability.GUIDED_DECODING the pipeline renders it onto the engine's native structured-output parameters in place of the control's live processor. The default returns None, which keeps the control on the live processor mechanism.

Parameters:

Name	Type	Description	Default
runtime_kwargs	dict | None	Per-call parameters supplied to generate().	None

Returns:

Type	Description
ConstraintSource | None	The constraint source, or None.

Source code in steerability/algorithms/output_control/base.py

def export_constraint(self, runtime_kwargs: dict | None = None) -> ConstraintSource | None:
    """The control's declarative constrained-decoding source, or None.

    A control whose per-step masking compiles from a declarative source returns a
    `ConstraintSource`; on a backend advertising `Capability.GUIDED_DECODING` the pipeline
    renders it onto the engine's native structured-output parameters in place of the
    control's live processor. The default returns None, which keeps the control on the live
    processor mechanism.

    Args:
        runtime_kwargs: Per-call parameters supplied to `generate()`.

    Returns:
        The constraint source, or None.
    """
    return None

export_generation_params(runtime_kwargs=None)

The configured stops as normalized generation parameters.

Source code in steerability/algorithms/output_control/stopping_rules/control.py

def export_generation_params(self, runtime_kwargs: dict | None = None) -> Mapping[str, Any]:
    """The configured stops as normalized generation parameters."""
    contribution: dict[str, Any] = {}
    if self.stop_texts:
        contribution["stop_strings"] = tuple(self.stop_texts)
    if self.stop_token_ids:
        contribution["stop_token_ids"] = tuple(self.stop_token_ids)
    if self.budget is not None:
        contribution["max_new_tokens"] = self.budget
    return contribution

export_processor_spec(runtime_kwargs=None)

The control's engine-hosted processor form, or None.

A control whose per-step logit math is expressible in an engine's served processor vocabulary returns a ProcessorSpec; on a backend advertising Capability.PER_STEP_LOGIT_SPECS with the spec's kind, the pipeline submits it as a ProcessorSpecEntry in place of the control's live processor. The default returns None, which keeps the control on the live processor mechanism.

Parameters:

Name	Type	Description	Default
runtime_kwargs	dict | None	Per-call parameters supplied to generate().	None

Returns:

Type	Description
ProcessorSpec | None	The processor spec, or None.

Source code in steerability/algorithms/output_control/base.py

def export_processor_spec(self, runtime_kwargs: dict | None = None) -> ProcessorSpec | None:
    """The control's engine-hosted processor form, or None.

    A control whose per-step logit math is expressible in an engine's served processor
    vocabulary returns a `ProcessorSpec`; on a backend advertising
    `Capability.PER_STEP_LOGIT_SPECS` with the spec's kind, the pipeline submits it as a
    `ProcessorSpecEntry` in place of the control's live processor. The default returns
    None, which keeps the control on the live processor mechanism.

    Args:
        runtime_kwargs: Per-call parameters supplied to `generate()`.

    Returns:
        The processor spec, or None.
    """
    return None

export_state()

Steer-time state to persist when freezing, keyed by logical name.

Values are typed artifacts (SteeringVector, Probe, ProbeSet, a Memory implementation, CheckpointArtifact, LoRAArtifact, torch.Tensor, or a Path to an on-disk product). Must be called after steer(). The default returns an empty mapping (nothing to persist).

Returns:

Type	Description
dict[str, Any]	Mapping from logical name to artifact value.

Source code in steerability/algorithms/core/base_control.py

def export_state(self) -> dict[str, Any]:
    """Steer-time state to persist when freezing, keyed by logical name.

    Values are typed artifacts (`SteeringVector`, `Probe`, `ProbeSet`, a `Memory`
    implementation, `CheckpointArtifact`, `LoRAArtifact`, `torch.Tensor`, or a `Path` to an
    on-disk product). Must be called after `steer()`. The default returns an empty mapping
    (nothing to persist).

    Returns:
        Mapping from logical name to artifact value.
    """
    return {}

fit_identity()

The object whose canonical form defines this control's fit digest, or None.

The digest detects recipe edits that invalidate frozen artifacts, so the returned object should cover exactly the inputs a re-fit would consume (training data, fit specs) and exclude inert application parameters (strengths, multipliers). The default returns None (no fits).

Returns:

Type	Description
Any | None	The fit-identity object, or None.

Source code in steerability/algorithms/core/base_control.py

def fit_identity(self) -> Any | None:
    """The object whose canonical form defines this control's fit digest, or None.

    The digest detects recipe edits that invalidate frozen artifacts, so the returned
    object should cover exactly the inputs a re-fit would consume (training data, fit
    specs) and exclude inert application parameters (strengths, multipliers). The default
    returns None (no fits).

    Returns:
        The fit-identity object, or None.
    """
    return None

frozen_form(state)

The (registry method key, constructor kwargs) of this control's frozen form.

state is the output of export_state(); artifact values inside the returned kwargs are replaced by store references at encode time. When steer_fits() is empty and state is empty, the recipe is the frozen form and the control's own key and recipe args are returned unchanged. A control whose freeze expands into several entries may return a list of (method key, kwargs) pairs instead.

Parameters:

Name	Type	Description	Default
state	dict[str, Any]	The output of export_state().	required

Returns:

Type	Description
tuple[str, dict[str, Any]]	The frozen method key and its constructor kwargs.

Raises:

Type	Description
NotFreezableError	If the control produces fits or state but declares no frozen form.

Source code in steerability/algorithms/core/base_control.py

def frozen_form(self, state: dict[str, Any]) -> tuple[str, dict[str, Any]]:
    """The `(registry method key, constructor kwargs)` of this control's frozen form.

    `state` is the output of `export_state()`; artifact values inside the returned kwargs
    are replaced by store references at encode time. When `steer_fits()` is empty and
    `state` is empty, the recipe is the frozen form and the control's own key and recipe
    args are returned unchanged. A control whose freeze expands into several entries may
    return a list of `(method key, kwargs)` pairs instead.

    Args:
        state: The output of `export_state()`.

    Returns:
        The frozen method key and its constructor kwargs.

    Raises:
        NotFreezableError: If the control produces fits or state but declares no frozen
            form.
    """
    if not state and not self.steer_fits():
        from steerability.algorithms.core.registry import method_key_for

        args = getattr(self, "args", None)
        kwargs = {field.name: getattr(args, field.name) for field in fields(args) if field.init} \
            if args is not None else {}
        return method_key_for(type(self)), kwargs
    raise NotFreezableError(
        f"{type(self).__name__} produces steer-time state but declares no frozen form; "
        "override frozen_form() (and export_state()) to support freezing."
    )

get_logits_processors(input_ids, runtime_kwargs, **kwargs)

The control's logits processors for the current generation.

Called once per generate() / compute_logprobs() call, after input and state controls have prepared the prompt (mirrors StateControl.get_hooks). **kwargs carries attention_mask and the caller's generation kwargs. Returned objects follow the HF LogitsProcessor convention; in-list order is preserved by the composition.

A processor must behave as a function of (prefix_ids, scores). Internal state is permitted only as memoization keyed on the prefix and must re-derive on a prefix mismatch, since drivers may restart, rewind, or reorder sequences, and scoring replays prefixes teacher-forced (subclass common.processors.base.PrefixKeyedProcessor to satisfy this mechanically). Return fresh processor instances from this hook; it is invoked once per call precisely so that per-generation state is isolated.

Parameters:

Name	Type	Description	Default
input_ids	Tensor	The steered prompt token ids [batch, seq_len].	required
runtime_kwargs	dict | None	Per-call parameters supplied to generate().	required

Returns:

Type	Description
list	A list of HF LogitsProcessor-style objects.

Source code in steerability/algorithms/output_control/base.py

def get_logits_processors(self, input_ids: torch.Tensor, runtime_kwargs: dict | None, **kwargs) -> list:
    """The control's logits processors for the current generation.

    Called once per `generate()` / `compute_logprobs()` call, after input and state
    controls have prepared the prompt (mirrors `StateControl.get_hooks`). `**kwargs`
    carries `attention_mask` and the caller's generation kwargs. Returned objects
    follow the HF `LogitsProcessor` convention; in-list order is preserved by the composition.

    A processor must behave as a function of `(prefix_ids, scores)`. Internal state is
    permitted only as memoization keyed on the prefix and must re-derive on a prefix mismatch,
    since drivers may restart, rewind, or reorder sequences, and scoring replays prefixes
    teacher-forced (subclass `common.processors.base.PrefixKeyedProcessor` to satisfy this
    mechanically). Return fresh processor instances from this hook; it is invoked once per call
    precisely so that per-generation state is isolated.

    Args:
        input_ids: The steered prompt token ids `[batch, seq_len]`.
        runtime_kwargs: Per-call parameters supplied to `generate()`.

    Returns:
        A list of HF `LogitsProcessor`-style objects.
    """
    return []

get_stopping_criteria(input_ids, runtime_kwargs, **kwargs)

Return fresh criteria anchored at the current prompt length.

Source code in steerability/algorithms/output_control/stopping_rules/control.py

def get_stopping_criteria(self, input_ids, runtime_kwargs, **kwargs) -> list:
    """Return fresh criteria anchored at the current prompt length."""
    prompt_len = input_ids.size(1)
    criteria = []
    for text in self.stop_texts:
        criteria.append(StopOnSubstring(self.tokenizer, text, prompt_len))
    if self.stop_token_ids:
        criteria.append(StopOnTokens(self.stop_token_ids))
    if self.budget is not None:
        criteria.append(BudgetTokens(self.budget, prompt_len))
    return criteria

requirements()

Stop rules are session contract on every backend, so no phase requires anything.

Source code in steerability/algorithms/output_control/stopping_rules/control.py

def requirements(self) -> Requirements:
    """Stop rules are session contract on every backend, so no phase requires anything."""
    return Requirements()

steer(model, tokenizer=None, **_)

Attach the tokenizer (required whenever stop_texts is configured).

Source code in steerability/algorithms/output_control/stopping_rules/control.py

def steer(self, model: PreTrainedModel, tokenizer: PreTrainedTokenizerBase | None = None, **_) -> PreTrainedModel:
    """Attach the tokenizer (required whenever `stop_texts` is configured)."""
    self.tokenizer = tokenizer or getattr(model, "tokenizer", None)
    if self.stop_texts and self.tokenizer is None:
        raise RuntimeError("StoppingRules requires a tokenizer when 'stop_texts' is configured.")
    return model

steer_access()

The model access this instance's steer step requires, on the ModelAccess ladder.

The default is ModelAccess.FACTS (layout and tokenizer only). A control whose steer step generates or scores through the session declares ROLLOUTS, one that captures hidden states declares CAPTURE, and one that touches the model as a live torch.nn.Module declares MODULE. Configuration determines the result. The pipeline hands steer() a session scoped to the declared rung, and the live model only at MODULE. A control may retain the pipeline model beyond steer() only if its generate phase requires Capability.IN_PROCESS_TORCH.

Returns:

Type	Description
ModelAccess	The declared access rung.

Source code in steerability/algorithms/core/base_control.py

def steer_access(self) -> ModelAccess:
    """The model access this instance's steer step requires, on the `ModelAccess` ladder.

    The default is `ModelAccess.FACTS` (layout and tokenizer only). A control whose steer
    step generates or scores through the session declares `ROLLOUTS`, one that captures
    hidden states declares `CAPTURE`, and one that touches the model as a live
    `torch.nn.Module` declares `MODULE`. Configuration determines the result. The pipeline
    hands `steer()` a session scoped to the declared rung, and the live model only at
    `MODULE`. A control may retain the pipeline model beyond `steer()` only if its
    generate phase requires `Capability.IN_PROCESS_TORCH`.

    Returns:
        The declared access rung.
    """
    return ModelAccess.FACTS

steer_fits()

The fit artifacts this instance's steer step will produce, for the steer plan.

Each entry is (artifact, artifact_class), where artifact is the fit source or recipe class name and artifact_class is "direction" or "calibrated". The default is an empty tuple (no fits).

Returns:

Type	Description
tuple[tuple[str, str], ...]	The declared fit artifacts, in declaration order.

Source code in steerability/algorithms/core/base_control.py

def steer_fits(self) -> tuple[tuple[str, str], ...]:
    """The fit artifacts this instance's steer step will produce, for the steer plan.

    Each entry is `(artifact, artifact_class)`, where `artifact` is the fit source or
    recipe class name and `artifact_class` is `"direction"` or `"calibrated"`. The default
    is an empty tuple (no fits).

    Returns:
        The declared fit artifacts, in declaration order.
    """
    return ()