Skip to content

Structural control

steerability.algorithms.structural_control.base

Structural control base classes.

This module provides the abstract base class for methods that create persistent changes to the model, either through weight updates or architectural changes.

Two base classes are provided:

  • StructuralControl: Base class for all structural control methods.

Structural controls implement steering through model weight or architecture modifications, transforming base parameters θ to θ', resulting in generations following y ~ p_θ'(x).

Examples of structural controls:

  • Fine-tuning (full or parameter-efficient like LoRA)
  • Model merging (e.g., via MergeKit)
  • Direct Preference Optimization (DPO)
  • Adapter layers and modules
  • Weight interpolation and averaging

See Also:

  • steerability.algorithms.structural_control: Implementations of structural control methods
  • steerability.core.steering_pipeline: Integration with steering pipeline

StructuralControl

Bases: BaseControl

Abstract base class for structural control steering methods.

Modifies model parameters or architecture persistently, returning a new model instance with transformed weights.

Methods:

Name Description
steer

Training logic (required)
Source code in steerability/algorithms/structural_control/base.py 
 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
class StructuralControl(BaseControl):
    """Abstract base class for structural control steering methods.

    Modifies model parameters or architecture persistently, returning a new model instance with transformed weights.

    Methods:
        steer(model, tokenizer, **kwargs) -> PreTrainedModel: Training logic (required)
    """

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

    enabled: bool = True
    supports_batching: bool = True

    @abstractmethod
    def steer(
            self,
            model: PreTrainedModel,
            tokenizer: PreTrainedTokenizerBase = None,
            session=None,
            **kwargs
    ) -> PreTrainedModel:
        """Required steering/preparation.

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

    def artifact_capability(self) -> Capability | None:
        """The serve capability implied by this configuration's steer-time artifact, or None.

        Controls whose `steer()` writes a servable product to disk return
        `Capability.SERVE_CHECKPOINT` for a full-weights checkpoint or `Capability.SERVE_LORA`
        for an adapter, so the generate phase gains a serving alternative. The default returns
        None (no on-disk artifact), which keeps the generate phase in-process only.

        Returns:
            The capability, or None.
        """
        return None

    def export_artifact(self) -> Artifact | None:
        """The steer-time artifact this control produced, or None.

        Called by the pipeline after `steer()` completes. The returned artifact must exist on
        disk and correspond to `artifact_capability()` (a `CheckpointArtifact` for
        `Capability.SERVE_CHECKPOINT`, a `LoRAArtifact` for `Capability.SERVE_LORA`). The
        default returns None.

        Returns:
            The artifact, or None.
        """
        return None

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

        The generate phase requires `Capability.IN_PROCESS_TORCH` for in-process adoption of
        the returned model; when the configuration produces an on-disk artifact
        (`artifact_capability()`), serving that artifact is an alternative, so a backend
        advertising the matching serve capability also supports the generate phase.

        Returns:
            The control's phase-keyed requirements.
        """
        generate = needs(Capability.IN_PROCESS_TORCH)
        capability = self.artifact_capability()
        if capability is not None:
            generate = any_of(
                generate,
                needs(capability, hint="serve the steer-time artifact on a vLLM backend"),
            )
        return Requirements(generate=generate)

    def steer_access(self) -> ModelAccess:
        """`ModelAccess.MODULE`; training happens on live weights."""
        return ModelAccess.MODULE

RUNTIME_KWARGS_SCHEMA = [] class-attribute instance-attribute

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

enabled = True class-attribute instance-attribute

supports_batching = True class-attribute instance-attribute

artifact_capability()

The serve capability implied by this configuration's steer-time artifact, or None.

Controls whose steer() writes a servable product to disk return Capability.SERVE_CHECKPOINT for a full-weights checkpoint or Capability.SERVE_LORA for an adapter, so the generate phase gains a serving alternative. The default returns None (no on-disk artifact), which keeps the generate phase in-process only.

Returns:

Type Description
Capability | None

The capability, or None.
Source code in steerability/algorithms/structural_control/base.py 
66
67
68
69
70
71
72
73
74
75
76
77
def artifact_capability(self) -> Capability | None:
    """The serve capability implied by this configuration's steer-time artifact, or None.

    Controls whose `steer()` writes a servable product to disk return
    `Capability.SERVE_CHECKPOINT` for a full-weights checkpoint or `Capability.SERVE_LORA`
    for an adapter, so the generate phase gains a serving alternative. The default returns
    None (no on-disk artifact), which keeps the generate phase in-process only.

    Returns:
        The capability, or None.
    """
    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_artifact()

The steer-time artifact this control produced, or None.

Called by the pipeline after steer() completes. The returned artifact must exist on disk and correspond to artifact_capability() (a CheckpointArtifact for Capability.SERVE_CHECKPOINT, a LoRAArtifact for Capability.SERVE_LORA). The default returns None.

Returns:

Type Description
Artifact | None

The artifact, or None.
Source code in steerability/algorithms/structural_control/base.py 
79
80
81
82
83
84
85
86
87
88
89
90
def export_artifact(self) -> Artifact | None:
    """The steer-time artifact this control produced, or None.

    Called by the pipeline after `steer()` completes. The returned artifact must exist on
    disk and correspond to `artifact_capability()` (a `CheckpointArtifact` for
    `Capability.SERVE_CHECKPOINT`, a `LoRAArtifact` for `Capability.SERVE_LORA`). The
    default returns None.

    Returns:
        The artifact, 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 
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.

The generate phase requires Capability.IN_PROCESS_TORCH for in-process adoption of the returned model; when the configuration produces an on-disk artifact (artifact_capability()), serving that artifact is an alternative, so a backend advertising the matching serve capability also supports the generate phase.

Returns:

Type Description
Requirements

The control's phase-keyed requirements.
Source code in steerability/algorithms/structural_control/base.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def requirements(self) -> Requirements:
    """Backend requirements computed from this instance's configuration, per phase.

    The generate phase requires `Capability.IN_PROCESS_TORCH` for in-process adoption of
    the returned model; when the configuration produces an on-disk artifact
    (`artifact_capability()`), serving that artifact is an alternative, so a backend
    advertising the matching serve capability also supports the generate phase.

    Returns:
        The control's phase-keyed requirements.
    """
    generate = needs(Capability.IN_PROCESS_TORCH)
    capability = self.artifact_capability()
    if capability is not None:
        generate = any_of(
            generate,
            needs(capability, hint="serve the steer-time artifact on a vLLM backend"),
        )
    return Requirements(generate=generate)

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

Required steering/preparation.

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

Source code in steerability/algorithms/structural_control/base.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
@abstractmethod
def steer(
        self,
        model: PreTrainedModel,
        tokenizer: PreTrainedTokenizerBase = None,
        session=None,
        **kwargs
) -> PreTrainedModel:
    """Required steering/preparation.

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

steer_access()

ModelAccess.MODULE; training happens on live weights.

Source code in steerability/algorithms/structural_control/base.py 
112
113
114
def steer_access(self) -> ModelAccess:
    """`ModelAccess.MODULE`; training happens on live weights."""
    return ModelAccess.MODULE

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