Skip to content

LoadLoRA

steerability.algorithms.structural_control.load_lora

args

LoadLoRA argument validation.

control

LoadLoRA: a structural control that attaches a saved LoRA adapter to the pipeline model.

logger = logging.getLogger(__name__) module-attribute

LoadLoRA

Bases: StructuralControl

Attaches a saved LoRA adapter to the pipeline model.

On the in-process backend, steer() verifies the adapter's base_model against the incoming model's reference, attaches the adapter with peft.PeftModel.from_pretrained, and merges it into the base weights when merge=True. On vLLM backends the adapter is served directly (Capability.SERVE_LORA), exported as a LoRAArtifact. The frozen form of an adapter-producing structural control (e.g. an SFT LoRA run) is an instance of this control pointing at the trained adapter.

With merge=False (the default) the pipeline model stays a PeftModel, so a state control listed after this control hooks the adapted model: layout resolution peels the PEFT wrapper and the hooked modules are the adapter's own.

Source code in steerability/algorithms/structural_control/load_lora/control.py 
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
class LoadLoRA(StructuralControl):
    """Attaches a saved LoRA adapter to the pipeline model.

    On the in-process backend, `steer()` verifies the adapter's `base_model` against the
    incoming model's reference, attaches the adapter with `peft.PeftModel.from_pretrained`,
    and merges it into the base weights when `merge=True`. On vLLM backends the adapter is
    served directly (`Capability.SERVE_LORA`), exported as a `LoRAArtifact`. The frozen form
    of an adapter-producing structural control (e.g. an SFT LoRA run) is an instance of this
    control pointing at the trained adapter.

    With `merge=False` (the default) the pipeline model stays a `PeftModel`, so a state control
    listed after this control hooks the adapted model: layout resolution peels the PEFT wrapper
    and the hooked modules are the adapter's own.
    """

    Args = LoadLoRAArgs
    supports_batching = True

    def artifact_capability(self) -> Capability:
        """`Capability.SERVE_LORA`; the configuration is an on-disk adapter."""
        return Capability.SERVE_LORA

    def export_artifact(self) -> LoRAArtifact:
        """The configured adapter directory and its base model reference."""
        return LoRAArtifact(path=str(self.path), base_model=str(self.base_model))

    def steer(
        self,
        model: PreTrainedModel,
        tokenizer: PreTrainedTokenizerBase | None = None,
        **kwargs,
    ) -> PreTrainedModel:
        """Attach the adapter to `model` and return the adapted model.

        Args:
            model: The pipeline model the adapter applies to.
            tokenizer: The pipeline tokenizer (unused).

        Returns:
            The adapted model (merged into the base weights when `merge=True`).

        Raises:
            ValueError: If `model` is None, or its reference differs from `base_model` and
                `allow_base_mismatch` is False.
        """
        from peft import PeftModel

        if model is None:
            raise ValueError("LoadLoRA requires the pipeline model; provide a base model.")
        if not self.allow_base_mismatch:
            live_ref = getattr(model, "name_or_path", None) or getattr(
                getattr(model, "config", None), "_name_or_path", None
            )
            if live_ref is not None and str(live_ref) != str(self.base_model):
                raise ValueError(
                    f"LoRA adapter was trained on base model {self.base_model!r} but the "
                    f"pipeline model is {live_ref!r}; load the matching base model, or set "
                    "allow_base_mismatch=True."
                )
        logger.info("Attaching LoRA adapter %s.", self.path)
        adapted = PeftModel.from_pretrained(model, self.path)
        if self.merge:
            adapted = adapted.merge_and_unload()
        return adapted
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
__init__(*args, **kwargs)
Source code in steerability/algorithms/core/base_control.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def __init__(self, *args, **kwargs) -> None:
    if self.Args is None:  # null / arg-free control
        if args or kwargs:
            raise TypeError(f"{type(self).__name__} accepts no constructor arguments.")
        self._configure()
        return

    self.args: BaseArgs = self.Args.validate(*args, **kwargs)

    # move fields to attributes, skipping any name the control exposes as a property
    # (e.g. CAST.condition_point); the raw value stays reachable via self.args.<name>
    for field in fields(self.args):
        if isinstance(getattr(type(self), field.name, None), property):
            continue
        setattr(self, field.name, getattr(self.args, field.name))

    self._configure()
_configure()

Post-construction hook, called after Args fields are mirrored onto the instance.

Driver presets override this to map their mirrored args onto the fields their generic base reads, so subclasses never bypass a parent __init__. Default no-op.

Source code in steerability/algorithms/core/base_control.py 
67
68
69
70
71
72
73
def _configure(self) -> None:
    """Post-construction hook, called after `Args` fields are mirrored onto the instance.

    Driver presets override this to map their mirrored args onto the fields their generic base
    reads, so subclasses never bypass a parent `__init__`. Default no-op.
    """
    pass
artifact_capability()

Capability.SERVE_LORA; the configuration is an on-disk adapter.

Source code in steerability/algorithms/structural_control/load_lora/control.py 
35
36
37
def artifact_capability(self) -> Capability:
    """`Capability.SERVE_LORA`; the configuration is an on-disk adapter."""
    return Capability.SERVE_LORA
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 configured adapter directory and its base model reference.

Source code in steerability/algorithms/structural_control/load_lora/control.py 
39
40
41
def export_artifact(self) -> LoRAArtifact:
    """The configured adapter directory and its base model reference."""
    return LoRAArtifact(path=str(self.path), base_model=str(self.base_model))
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, **kwargs)

Attach the adapter to model and return the adapted model.

Parameters:

Name Type Description Default
model PreTrainedModel

The pipeline model the adapter applies to.

 required
tokenizer PreTrainedTokenizerBase | None

The pipeline tokenizer (unused).

 None

Returns:

Type Description
PreTrainedModel

The adapted model (merged into the base weights when merge=True).

Raises:

Type Description
ValueError

If model is None, or its reference differs from base_model and allow_base_mismatch is False.
Source code in steerability/algorithms/structural_control/load_lora/control.py 
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
def steer(
    self,
    model: PreTrainedModel,
    tokenizer: PreTrainedTokenizerBase | None = None,
    **kwargs,
) -> PreTrainedModel:
    """Attach the adapter to `model` and return the adapted model.

    Args:
        model: The pipeline model the adapter applies to.
        tokenizer: The pipeline tokenizer (unused).

    Returns:
        The adapted model (merged into the base weights when `merge=True`).

    Raises:
        ValueError: If `model` is None, or its reference differs from `base_model` and
            `allow_base_mismatch` is False.
    """
    from peft import PeftModel

    if model is None:
        raise ValueError("LoadLoRA requires the pipeline model; provide a base model.")
    if not self.allow_base_mismatch:
        live_ref = getattr(model, "name_or_path", None) or getattr(
            getattr(model, "config", None), "_name_or_path", None
        )
        if live_ref is not None and str(live_ref) != str(self.base_model):
            raise ValueError(
                f"LoRA adapter was trained on base model {self.base_model!r} but the "
                f"pipeline model is {live_ref!r}; load the matching base model, or set "
                "allow_base_mismatch=True."
            )
    logger.info("Attaching LoRA adapter %s.", self.path)
    adapted = PeftModel.from_pretrained(model, self.path)
    if self.merge:
        adapted = adapted.merge_and_unload()
    return adapted
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 ()