Skip to content

Input control

steerability.algorithms.input_control.base

Input control base classes.

This module provides the abstract base class for methods that modify prompts before they reach the model.

Two base classes are provided:

  • InputControl: Base class for all input control methods.

Input controls implement steering through prompt transformation σ(x), enabling behavior modification without altering model parameters or architecture. These methods transform inputs before they reach the model, resulting in generations following y ~ p_θ(σ(x)).

Examples of input controls:

  • Few-shot learning (prepending examples)
  • Prompt templates and formatting
  • Soft prompts and prompt tuning
  • Chain-of-thought prompting
  • Iterative prompt refinement

See Also:

  • steerability.algorithms.input_control: Implementations of input control methods
  • steerability.core.steering_pipeline: Integration with steering pipeline

InputControl

Bases: BaseControl

Abstract base class for input control steering methods.

Transforms a prompt before it reaches the model, steering behavior via the input alone (no changes to weights, architecture, or runtime activations). Any preparation happens once in steer(); adapt() / adapt_messages() are a function of the input and the prepared state at inference time.

A pipeline may hold several input controls; they chain in controls-list order, each control receiving the previous control's output within its phase (message-level for chat input, then token-level).

Methods:

Name Description
adapt

Tensor-level adaptation (required).
adapt_messages

Optional message-level adaptation, called BEFORE chat-template tokenization. Default returns None (no change).
steer

One-time preparation (optional).
cleanup

Release resources allocated during steer (optional).

Subclasses that produce an artifact in steer() (instructions, demonstrations, learned weights, ...) may expose it via the memory attribute, e.g., see TextMemory.

Source code in steerability/algorithms/input_control/base.py 
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
class InputControl(BaseControl):
    """Abstract base class for input control steering methods.

    Transforms a prompt before it reaches the model, steering behavior via the input alone (no changes to weights,
    architecture, or runtime activations). Any preparation happens once in `steer()`; `adapt()` / `adapt_messages()`
    are a function of the input and the prepared state at inference time.

    A pipeline may hold several input controls; they chain in `controls`-list order, each control receiving the
    previous control's output within its phase (message-level for chat input, then token-level).

    Methods:
        adapt(input_ids, runtime_kwargs) -> input_ids: Tensor-level adaptation (required).
        adapt_messages(messages, runtime_kwargs) -> messages | None: Optional message-level adaptation,
            called BEFORE chat-template tokenization. Default returns None (no change).
        steer(model, tokenizer, **kwargs) -> None: One-time preparation (optional).
        cleanup() -> None: Release resources allocated during steer (optional).

    Subclasses that produce an artifact in `steer()` (instructions, demonstrations, learned weights, ...) may expose it
    via the `memory` attribute, e.g., see `TextMemory`.
    """

    Args: type[BaseArgs] | None = None
    RUNTIME_KWARGS_SCHEMA: list[dict] = []

    enabled: bool = True
    supports_batching: bool = False

    memory: Memory | None = None  # subclasses populate in steer()

    @abstractmethod
    def adapt(
        self,
        input_ids: list[int] | torch.Tensor,
        runtime_kwargs: dict | None = None,
    ) -> list[int] | torch.Tensor:
        """Transform `input_ids` into a steered prompt.

        May read instance state (e.g. `self.memory`) that was populated by `steer()`.

        Args:
            input_ids: The user's prompt token IDs.
            runtime_kwargs: Per-call parameters.

        Returns:
            The transformed token IDs.
        """

    def adapt_messages(
        self,
        messages: list[list[dict]],
        runtime_kwargs: dict | None = None,
    ) -> list[list[dict]] | None:
        """Optional message-level adaptation, called BEFORE chat-template tokenization.

        Default returns None (no message-level changes). Subclasses that modify chat structure (set/replace system
        prompt, insert example turns) override this. When this method returns a non-None result for a chat-input
        generation, the pipeline does NOT additionally call this control's `adapt()` for that call; a control may
        therefore implement both entry points (message-level for chat input, token-level for raw text/tensor input)
        without being applied twice. Other input controls in the pipeline still run per their own phase.

        Args:
            messages: A batch of chats; outer list is the batch, inner list is the message sequence for one chat.
            runtime_kwargs: Per-call parameters.

        Returns:
            The transformed messages, or None to indicate no change.
        """
        return None

    def steer(
        self,
        model=None,
        tokenizer=None,
        session=None,
        **kwargs,
    ) -> None:
        """Optional offline preparation. Default is no-op.

        `session` is a `SteeringSession` on the steering backend, provided by the pipeline.
        """
        pass

    def requirements(self) -> Requirements:
        """Backend requirements computed from this instance's configuration, per phase.

        Input controls transform the prompt client-side, so the generate phase requires nothing
        beyond the session contract on any backend. A control whose `steer()` reads the live
        pipeline model (e.g. for rollouts or scoring) overrides this with a steer-phase
        `Capability.IN_PROCESS_TORCH` requirement.

        Returns:
            The control's phase-keyed requirements.
        """
        return Requirements()

RUNTIME_KWARGS_SCHEMA = [] class-attribute instance-attribute

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

enabled = True class-attribute instance-attribute

memory = None class-attribute instance-attribute

supports_batching = False class-attribute instance-attribute

adapt(input_ids, runtime_kwargs=None) abstractmethod

Transform input_ids into a steered prompt.

May read instance state (e.g. self.memory) that was populated by steer().

Parameters:

Name Type Description Default
input_ids list[int] | Tensor

The user's prompt token IDs.

 required
runtime_kwargs dict | None

Per-call parameters.

 None

Returns:

Type Description
list[int] | Tensor

The transformed token IDs.
Source code in steerability/algorithms/input_control/base.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@abstractmethod
def adapt(
    self,
    input_ids: list[int] | torch.Tensor,
    runtime_kwargs: dict | None = None,
) -> list[int] | torch.Tensor:
    """Transform `input_ids` into a steered prompt.

    May read instance state (e.g. `self.memory`) that was populated by `steer()`.

    Args:
        input_ids: The user's prompt token IDs.
        runtime_kwargs: Per-call parameters.

    Returns:
        The transformed token IDs.
    """

adapt_messages(messages, runtime_kwargs=None)

Optional message-level adaptation, called BEFORE chat-template tokenization.

Default returns None (no message-level changes). Subclasses that modify chat structure (set/replace system prompt, insert example turns) override this. When this method returns a non-None result for a chat-input generation, the pipeline does NOT additionally call this control's adapt() for that call; a control may therefore implement both entry points (message-level for chat input, token-level for raw text/tensor input) without being applied twice. Other input controls in the pipeline still run per their own phase.

Parameters:

Name Type Description Default
messages list[list[dict]]

A batch of chats; outer list is the batch, inner list is the message sequence for one chat.

 required
runtime_kwargs dict | None

Per-call parameters.

 None

Returns:

Type Description
list[list[dict]] | None

The transformed messages, or None to indicate no change.
Source code in steerability/algorithms/input_control/base.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def adapt_messages(
    self,
    messages: list[list[dict]],
    runtime_kwargs: dict | None = None,
) -> list[list[dict]] | None:
    """Optional message-level adaptation, called BEFORE chat-template tokenization.

    Default returns None (no message-level changes). Subclasses that modify chat structure (set/replace system
    prompt, insert example turns) override this. When this method returns a non-None result for a chat-input
    generation, the pipeline does NOT additionally call this control's `adapt()` for that call; a control may
    therefore implement both entry points (message-level for chat input, token-level for raw text/tensor input)
    without being applied twice. Other input controls in the pipeline still run per their own phase.

    Args:
        messages: A batch of chats; outer list is the batch, inner list is the message sequence for one chat.
        runtime_kwargs: Per-call parameters.

    Returns:
        The transformed messages, or None to indicate no change.
    """
    return None

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 
195
196
197
198
199
200
201
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 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
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_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 
117
118
119
120
121
122
123
124
125
126
127
128
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 
161
162
163
164
165
166
167
168
169
170
171
172
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 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
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."
    )

requirements()

Backend requirements computed from this instance's configuration, per phase.

Input controls transform the prompt client-side, so the generate phase requires nothing beyond the session contract on any backend. A control whose steer() reads the live pipeline model (e.g. for rollouts or scoring) overrides this with a steer-phase Capability.IN_PROCESS_TORCH requirement.

Returns:

Type Description
Requirements

The control's phase-keyed requirements.
Source code in steerability/algorithms/input_control/base.py 
124
125
126
127
128
129
130
131
132
133
134
135
def requirements(self) -> Requirements:
    """Backend requirements computed from this instance's configuration, per phase.

    Input controls transform the prompt client-side, so the generate phase requires nothing
    beyond the session contract on any backend. A control whose `steer()` reads the live
    pipeline model (e.g. for rollouts or scoring) overrides this with a steer-phase
    `Capability.IN_PROCESS_TORCH` requirement.

    Returns:
        The control's phase-keyed requirements.
    """
    return Requirements()

steer(model=None, tokenizer=None, session=None, **kwargs)

Optional offline preparation. Default is no-op.

session is a SteeringSession on the steering backend, provided by the pipeline.

Source code in steerability/algorithms/input_control/base.py 
111
112
113
114
115
116
117
118
119
120
121
122
def steer(
    self,
    model=None,
    tokenizer=None,
    session=None,
    **kwargs,
) -> None:
    """Optional offline preparation. Default is no-op.

    `session` is a `SteeringSession` on the steering backend, provided by the pipeline.
    """
    pass

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 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
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 
105
106
107
108
109
110
111
112
113
114
115
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 ()