Skip to content

SystemPrompt

steerability.algorithms.input_control.system_prompt

args

control

Input control that sets or merges the leading system message of a chat.

SystemPrompt

Bases: InputControl

Set or merge the leading system message of a chat.

The text is a constant string configured at construction (no runtime kwargs). On chat input the leading system message (chat[0] when its role is "system") is edited according to mode: "prepend" places the text ahead of the existing content, "append" after it, and "replace" substitutes it, joining with separator for "prepend" and "append". When the chat has no leading system message, all three modes insert one carrying the text at position 0, so separator has no effect. Every input shape yields exactly one leading system message. System messages elsewhere in the chat are left untouched.

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 raw text or tensor input; it decodes, re-templates, and re-encodes, and because decoding flattens any system prompt into text it cannot merge, so mode does not apply there and every mode behaves as "replace". 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

System-prompt text set or merged into the leading system message. Non-empty.

 required
mode

How the text combines with an existing leading system message ("prepend" (default), "append", or "replace"). Ignored when no leading system message is present.

 required
separator

String inserted between the text and the existing content for "prepend" and "append". Empty string allowed.

 required
Source code in steerability/algorithms/input_control/system_prompt/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
144
145
146
147
148
149
150
151
class SystemPrompt(InputControl):
    """Set or merge the leading system message of a chat.

    The text is a constant string configured at construction (no runtime kwargs). On chat input the leading
    system message (`chat[0]` when its role is `"system"`) is edited according to `mode`: `"prepend"` places the
    text ahead of the existing content, `"append"` after it, and `"replace"` substitutes it, joining with
    `separator` for `"prepend"` and `"append"`. When the chat has no leading system message, all three modes
    insert one carrying the text at position 0, so `separator` has no effect. Every input shape yields exactly
    one leading system message. System messages elsewhere in the chat are left untouched.

    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 raw text or tensor
    input; it decodes, re-templates, and re-encodes, and because decoding flattens any system prompt into text it
    cannot merge, so `mode` does not apply there and every mode behaves as `"replace"`. 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: System-prompt text set or merged into the leading system message. Non-empty.
        mode: How the text combines with an existing leading system message (`"prepend"` (default), `"append"`,
            or `"replace"`). Ignored when no leading system message is present.
        separator: String inserted between the text and the existing content for `"prepend"` and `"append"`.
            Empty string allowed.
    """

    Args = SystemPromptArgs

    supports_batching: bool = True

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

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

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

    def adapt_messages(
        self,
        messages: list[list[dict]],
        runtime_kwargs: dict | None = None,
    ) -> list[list[dict]]:
        """Set or merge the leading system message 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, each with exactly one leading system message.
        """
        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:
        """Set the system prompt on the token stream.

        Fallback path for raw text or tensor input; the message phase is the faithful path for chat input. The
        stream is decoded, re-templated with the text as the system message, and re-encoded, so `mode` does not
        apply and every mode behaves as `"replace"`. 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 system prompt applied.

        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("SystemPrompt 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())

        max_len = max(len(seq) for seq in adapted_batch)
        if len(adapted_batch) > 1:
            if self.tokenizer.pad_token_id is None:
                raise RuntimeError(
                    "SystemPrompt: tokenizer has no pad_token_id; cannot pad batch sequences. "
                    "Set a pad token before using SystemPrompt with batched inputs."
                )
            pad_id = self.tokenizer.pad_token_id
            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
mode = 'prepend' 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)

Set the system prompt on the token stream.

Fallback path for raw text or tensor input; the message phase is the faithful path for chat input. The stream is decoded, re-templated with the text as the system message, and re-encoded, so mode does not apply and every mode behaves as "replace". 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 system prompt applied.

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/system_prompt/control.py 
 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
144
145
146
147
148
149
150
151
def adapt(
    self,
    input_ids: list[int] | torch.Tensor,
    runtime_kwargs: dict | None = None,
) -> list[int] | torch.Tensor:
    """Set the system prompt on the token stream.

    Fallback path for raw text or tensor input; the message phase is the faithful path for chat input. The
    stream is decoded, re-templated with the text as the system message, and re-encoded, so `mode` does not
    apply and every mode behaves as `"replace"`. 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 system prompt applied.

    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("SystemPrompt 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())

    max_len = max(len(seq) for seq in adapted_batch)
    if len(adapted_batch) > 1:
        if self.tokenizer.pad_token_id is None:
            raise RuntimeError(
                "SystemPrompt: tokenizer has no pad_token_id; cannot pad batch sequences. "
                "Set a pad token before using SystemPrompt with batched inputs."
            )
        pad_id = self.tokenizer.pad_token_id
        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)

Set or merge the leading system message 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, each with exactly one leading system message.
Source code in steerability/algorithms/input_control/system_prompt/control.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def adapt_messages(
    self,
    messages: list[list[dict]],
    runtime_kwargs: dict | None = None,
) -> list[list[dict]]:
    """Set or merge the leading system message 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, each with exactly one leading system message.
    """
    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/system_prompt/control.py 
52
53
54
55
56
57
58
59
60
def steer(
    self,
    model=None,
    tokenizer: PreTrainedTokenizerBase | None = None,
    **kwargs,
) -> None:
    self.tokenizer = tokenizer
    self.memory = TextMemory(slots={"instruction": self.text})
    self._formatter = SystemPromptFormatter(mode=self.mode, separator=self.separator)
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 ()