Skip to content

UserPrefix

steerability.algorithms.input_control.user_prefix

args

control

Input control that prepends a fixed text marker to a user turn.

UserPrefix

Bases: InputControl

Prepend a fixed text marker to a user turn.

The marker is a constant string configured at construction (no runtime kwargs). On chat input the marker is concatenated to the content of the user turn(s) selected by placement, joined by separator; when a chat has no user turn a new user turn carrying the marker is appended. On raw text or tensor input the encoded marker is prefixed to the token stream.

The control implements both adaptation phases. The message phase (adapt_messages) is the faithful path for chat input, applied before chat templating. The token phase (adapt) is the fallback for non-chat input and prefixes the encoded marker to the ids. The base-class contract guarantees no double application: a non-None return from adapt_messages skips this control's adapt for that call.

Parameters:

Name Type Description Default
text

Marker text prepended to the targeted user turn(s). Non-empty.

 required
separator

String inserted between the marker and the existing user content. Empty string allowed.

 required
placement

Which user turn(s) receive the marker ("last_user" (default), "first_user", or "all_user").

 required
Source code in steerability/algorithms/input_control/user_prefix/control.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 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
136
137
138
139
140
141
142
143
class UserPrefix(InputControl):
    """Prepend a fixed text marker to a user turn.

    The marker is a constant string configured at construction (no runtime kwargs). On chat input the marker is
    concatenated to the content of the user turn(s) selected by `placement`, joined by `separator`; when a chat
    has no user turn a new user turn carrying the marker is appended. On raw text or tensor input the encoded
    marker is prefixed to the token stream.

    The control implements both adaptation phases. The message phase (`adapt_messages`) is the faithful path for
    chat input, applied before chat templating. The token phase (`adapt`) is the fallback for non-chat input and
    prefixes the encoded marker to the ids. The base-class contract guarantees no double application: a non-`None`
    return from `adapt_messages` skips this control's `adapt` for that call.

    Args:
        text: Marker text prepended to the targeted user turn(s). Non-empty.
        separator: String inserted between the marker and the existing user content. Empty string allowed.
        placement: Which user turn(s) receive the marker (`"last_user"` (default), `"first_user"`, or `"all_user"`).
    """

    Args = UserPrefixArgs

    supports_batching: bool = True

    # placeholders (dataclass attrs from UserPrefixArgs override these at __init__ time)
    tokenizer: PreTrainedTokenizerBase | None = None
    text: str | None = None
    separator: str = "\n\n"
    placement: str = "last_user"

    # method-owned state populated in steer()
    memory: TextMemory | None = None
    _formatter: PrependTextFormatter | None = None

    def steer(
        self,
        model=None,
        tokenizer: PreTrainedTokenizerBase | None = None,
        **kwargs,
    ) -> None:
        self.tokenizer = tokenizer
        self.memory = TextMemory(slots={"text": self.text})
        self._formatter = PrependTextFormatter(separator=self.separator, target=self.placement)

    def adapt_messages(
        self,
        messages: list[list[dict]],
        runtime_kwargs: dict | None = None,
    ) -> list[list[dict]]:
        """Prepend the marker to the targeted user turn of each chat.

        Always returns the adapted batch (never `None`), since `text` is validated non-empty and the control
        therefore always changes chat input.

        Args:
            messages: A batch of chats; outer list is the batch, inner list is one chat's message sequence.
            runtime_kwargs: Unused.

        Returns:
            The adapted batch of chats.
        """
        return self._formatter.apply_to_messages(messages, self.memory)

    def adapt(
        self,
        input_ids: list[int] | torch.Tensor,
        runtime_kwargs: dict | None = None,
    ) -> list[int] | torch.Tensor:
        """Prefix the encoded marker to the token ids.

        Fallback path for raw text or tensor input; the message phase is the faithful path for chat input.
        Handles `list[int]`, `list[list[int]]`, and 1-D or 2-D tensors, preserving the input container, dtype, and
        device on output. Batched sequences are padded to a uniform length.

        Args:
            input_ids: The user's prompt token ids.
            runtime_kwargs: Unused.

        Returns:
            The token ids with the encoded marker prefixed.

        Raises:
            RuntimeError: If the tokenizer is not set (requires calling `steer()` first), or if a batch must be
                padded but the tokenizer has no `pad_token_id`.
        """
        if self.tokenizer is None:
            raise RuntimeError("UserPrefix needs a tokenizer; call .steer() first.")

        is_tensor = isinstance(input_ids, torch.Tensor)
        original_device = input_ids.device if is_tensor else None
        original_dtype = input_ids.dtype if is_tensor else None

        # normalize to 2D list format [batch_size, seq_len]
        if is_tensor:
            if input_ids.ndim == 1:
                batch_input_ids = [input_ids.tolist()]
                single_sequence = True
            else:
                batch_input_ids = input_ids.tolist()
                single_sequence = False
        else:
            if isinstance(input_ids[0], int):
                batch_input_ids = [input_ids]
                single_sequence = True
            else:
                batch_input_ids = input_ids
                single_sequence = False

        adapted_batch: list[list[int]] = []
        for input_ids_single in batch_input_ids:
            input_tensor = torch.tensor(input_ids_single, dtype=torch.long).unsqueeze(0)
            adapted_tensor = self._formatter.apply_to_ids(input_tensor, self.memory, self.tokenizer)
            adapted_batch.append(adapted_tensor[0].tolist())

        if len(adapted_batch) > 1:
            if self.tokenizer.pad_token_id is None:
                raise RuntimeError(
                    "UserPrefix: tokenizer has no pad_token_id; cannot pad batch sequences. "
                    "Set a pad token before using UserPrefix with batched inputs."
                )
            pad_id = self.tokenizer.pad_token_id
            max_len = max(len(seq) for seq in adapted_batch)
            adapted_batch = [seq + [pad_id] * (max_len - len(seq)) for seq in adapted_batch]

        if is_tensor:
            result = torch.tensor(adapted_batch, dtype=original_dtype, device=original_device)
            if single_sequence:
                result = result.squeeze(0)
            return result
        if single_sequence:
            return adapted_batch[0]
        return adapted_batch
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
placement = 'last_user' class-attribute instance-attribute
separator = '\n\n' class-attribute instance-attribute
supports_batching = True class-attribute instance-attribute
text = None class-attribute instance-attribute
tokenizer = None class-attribute instance-attribute
adapt(input_ids, runtime_kwargs=None)

Prefix the encoded marker to the token ids.

Fallback path for raw text or tensor input; the message phase is the faithful path for chat input. Handles list[int], list[list[int]], and 1-D or 2-D tensors, preserving the input container, dtype, and device on output. Batched sequences are padded to a uniform length.

Parameters:

Name Type Description Default
input_ids list[int] | Tensor

The user's prompt token ids.

 required
runtime_kwargs dict | None

Unused.

 None

Returns:

Type Description
list[int] | Tensor

The token ids with the encoded marker prefixed.

Raises:

Type Description
RuntimeError

If the tokenizer is not set (requires calling steer() first), or if a batch must be padded but the tokenizer has no pad_token_id.
Source code in steerability/algorithms/input_control/user_prefix/control.py 
 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
136
137
138
139
140
141
142
143
def adapt(
    self,
    input_ids: list[int] | torch.Tensor,
    runtime_kwargs: dict | None = None,
) -> list[int] | torch.Tensor:
    """Prefix the encoded marker to the token ids.

    Fallback path for raw text or tensor input; the message phase is the faithful path for chat input.
    Handles `list[int]`, `list[list[int]]`, and 1-D or 2-D tensors, preserving the input container, dtype, and
    device on output. Batched sequences are padded to a uniform length.

    Args:
        input_ids: The user's prompt token ids.
        runtime_kwargs: Unused.

    Returns:
        The token ids with the encoded marker prefixed.

    Raises:
        RuntimeError: If the tokenizer is not set (requires calling `steer()` first), or if a batch must be
            padded but the tokenizer has no `pad_token_id`.
    """
    if self.tokenizer is None:
        raise RuntimeError("UserPrefix needs a tokenizer; call .steer() first.")

    is_tensor = isinstance(input_ids, torch.Tensor)
    original_device = input_ids.device if is_tensor else None
    original_dtype = input_ids.dtype if is_tensor else None

    # normalize to 2D list format [batch_size, seq_len]
    if is_tensor:
        if input_ids.ndim == 1:
            batch_input_ids = [input_ids.tolist()]
            single_sequence = True
        else:
            batch_input_ids = input_ids.tolist()
            single_sequence = False
    else:
        if isinstance(input_ids[0], int):
            batch_input_ids = [input_ids]
            single_sequence = True
        else:
            batch_input_ids = input_ids
            single_sequence = False

    adapted_batch: list[list[int]] = []
    for input_ids_single in batch_input_ids:
        input_tensor = torch.tensor(input_ids_single, dtype=torch.long).unsqueeze(0)
        adapted_tensor = self._formatter.apply_to_ids(input_tensor, self.memory, self.tokenizer)
        adapted_batch.append(adapted_tensor[0].tolist())

    if len(adapted_batch) > 1:
        if self.tokenizer.pad_token_id is None:
            raise RuntimeError(
                "UserPrefix: tokenizer has no pad_token_id; cannot pad batch sequences. "
                "Set a pad token before using UserPrefix with batched inputs."
            )
        pad_id = self.tokenizer.pad_token_id
        max_len = max(len(seq) for seq in adapted_batch)
        adapted_batch = [seq + [pad_id] * (max_len - len(seq)) for seq in adapted_batch]

    if is_tensor:
        result = torch.tensor(adapted_batch, dtype=original_dtype, device=original_device)
        if single_sequence:
            result = result.squeeze(0)
        return result
    if single_sequence:
        return adapted_batch[0]
    return adapted_batch
adapt_messages(messages, runtime_kwargs=None)

Prepend the marker to the targeted user turn of each chat.

Always returns the adapted batch (never None), since text is validated non-empty and the control therefore always changes chat input.

Parameters:

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

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

 required
runtime_kwargs dict | None

Unused.

 None

Returns:

Type Description
list[list[dict]]

The adapted batch of chats.
Source code in steerability/algorithms/input_control/user_prefix/control.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def adapt_messages(
    self,
    messages: list[list[dict]],
    runtime_kwargs: dict | None = None,
) -> list[list[dict]]:
    """Prepend the marker to the targeted user turn of each chat.

    Always returns the adapted batch (never `None`), since `text` is validated non-empty and the control
    therefore always changes chat input.

    Args:
        messages: A batch of chats; outer list is the batch, inner list is one chat's message sequence.
        runtime_kwargs: Unused.

    Returns:
        The adapted batch of chats.
    """
    return self._formatter.apply_to_messages(messages, self.memory)
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, **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/user_prefix/control.py 
46
47
48
49
50
51
52
53
54
def steer(
    self,
    model=None,
    tokenizer: PreTrainedTokenizerBase | None = None,
    **kwargs,
) -> None:
    self.tokenizer = tokenizer
    self.memory = TextMemory(slots={"text": self.text})
    self._formatter = PrependTextFormatter(separator=self.separator, target=self.placement)
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 ()