Payload classes (vllm.steer_vectors)

Payloads describe weights independently of source file formats. Construct them from NumPy or PyTorch tensors, or obtain them through an easysteer.vectors adapter, then pass them as VectorSpec(data=payload, ...).

from vllm.steer_vectors import (
    ConceptPair,
    DirectionVector,
    LinearMap,
    LowRankProjector,
    Payload,
    ReftIntervention,
    RouterConfig,
)

The tensor payload constructors validate dimensions and finite values and store contiguous float32 arrays. Engine admission checks component widths and target layers; worker materialization uses the configured steering dtype. See algorithms and payloads for transformations and file schemas.

Common wire representation

All payload classes inherit Payload.to_wire(). It returns a dictionary with version, kind, named tensors (shape and raw byte data), extra metadata, and a content sha256. Use to_json_payload() for JSON/HTTP encoding of those bytes.

vllm.steer_vectors.Payload

Base class: named float32 tensors plus payload-specific fields.

Source code in vllm-steer/vllm/model_hooks/steering/payloads.py

class Payload:
    """Base class: named float32 tensors plus payload-specific fields."""

    kind: str  # set by subclasses

    def _tensors(self) -> dict[str, np.ndarray]:
        raise NotImplementedError

    def _extra(self) -> dict[str, Any]:
        return {}

    def to_wire(self) -> dict[str, Any]:
        tensors = self._tensors()
        wire_tensors = {}
        for name in sorted(tensors):
            arr = tensors[name]
            wire_tensors[name] = {
                "shape": list(arr.shape),
                "data": arr.tobytes(),
            }
        wire = {
            "version": _WIRE_VERSION,
            "kind": self.kind,
            "tensors": wire_tensors,
            "extra": copy.deepcopy(self._extra()),
        }
        wire["sha256"] = _wire_digest(wire)
        return wire

kind instance-attribute

kind: str

to_wire

to_wire() -> dict[str, Any]

Source code in vllm-steer/vllm/model_hooks/steering/payloads.py

def to_wire(self) -> dict[str, Any]:
    tensors = self._tensors()
    wire_tensors = {}
    for name in sorted(tensors):
        arr = tensors[name]
        wire_tensors[name] = {
            "shape": list(arr.shape),
            "data": arr.tobytes(),
        }
    wire = {
        "version": _WIRE_VERSION,
        "kind": self.kind,
        "tensors": wire_tensors,
        "extra": copy.deepcopy(self._extra()),
    }
    wire["sha256"] = _wire_digest(wire)
    return wire

Direction weights

layers maps true decoder-layer IDs to one-dimensional vectors. Widths may differ across layers when the model component does. For attention_add, each vector contains concatenated head outputs; obtain their dimensions from capture layouts. The constructor returns a DirectionVector; algorithms determine how its directions are applied.

vllm.steer_vectors.DirectionVector

DirectionVector(layers: dict[int, Any])

Bases: Payload

Per-layer direction vectors: h' = h + scale * v.

Accepted by direct, erase, replace and attention_add. Each true layer id maps to a 1-D vector; admission checks its width against the component.

Source code in vllm-steer/vllm/model_hooks/steering/payloads.py

def __init__(self, layers: dict[int, Any]):
    if not layers:
        raise ValueError("DirectionVector requires at least one layer")
    self.layers = {
        int(layer): _as_array(f"layers[{layer}]", vec, ndim=1)
        for layer, vec in layers.items()
    }

kind class-attribute instance-attribute

kind = 'direction'

layers instance-attribute

layers = {
    int(layer): _as_array(f"layers[{layer}]", vec, ndim=1)
    for layer, vec in layers.items()
}

Affine and low-rank maps

For LinearMap, weight has shape (hidden, hidden) and optional bias has shape (hidden,). For LowRankProjector, both projectors have shape (hidden, rank). These payloads do not record layer IDs; specify VectorSpec.layers when applying the returned object.

vllm.steer_vectors.LinearMap

LinearMap(weight: Any, bias: Any = None)

Bases: Payload

An affine map h' = h @ W.T + b applied to each target layer.

Accepted by the linear algorithm. Requires VectorSpec.layers.

Source code in vllm-steer/vllm/model_hooks/steering/payloads.py

def __init__(self, weight: Any, bias: Any = None):
    self.weight = _as_array("weight", weight, ndim=2)
    if self.weight.shape[0] != self.weight.shape[1]:
        raise ValueError(f"weight must be square, got shape {self.weight.shape}")
    self.bias = None if bias is None else _as_array("bias", bias, ndim=1)
    if self.bias is not None and self.bias.shape[0] != self.weight.shape[0]:
        raise ValueError(
            f"bias size {self.bias.shape[0]} != weight rows {self.weight.shape[0]}"
        )

kind class-attribute instance-attribute

kind = 'linear'

weight instance-attribute

weight = _as_array('weight', weight, ndim=2)

bias instance-attribute

bias = (
    None
    if bias is None
    else _as_array("bias", bias, ndim=1)
)

vllm.steer_vectors.LowRankProjector

LowRankProjector(projector1: Any, projector2: Any)

Bases: Payload

A low-rank update h' = h + scale * (h @ P1) @ P2.T.

Accepted by the lm_steer algorithm. Requires VectorSpec.layers.

Source code in vllm-steer/vllm/model_hooks/steering/payloads.py

def __init__(self, projector1: Any, projector2: Any):
    self.projector1 = _as_array("projector1", projector1, ndim=2)
    self.projector2 = _as_array("projector2", projector2, ndim=2)
    _check_dims(
        {"projector1": self.projector1, "projector2": self.projector2},
        {"projector1": 0, "projector2": 0},
    )
    if self.projector1.shape[1] != self.projector2.shape[1]:
        raise ValueError(
            f"projector ranks disagree: {self.projector1.shape[1]} vs "
            f"{self.projector2.shape[1]}"
        )

kind class-attribute instance-attribute

kind = 'lowrank'

projector1 instance-attribute

projector1 = _as_array('projector1', projector1, ndim=2)

projector2 instance-attribute

projector2 = _as_array('projector2', projector2, ndim=2)

LoReFT weights

rotate_layer has shape (hidden, rank), learned_source_weight has shape (rank, hidden), and optional learned_source_bias has shape (rank,). layer records a checkpoint's target. If it is omitted, supply VectorSpec.layers. The returned ReftIntervention is accepted by loreft.

vllm.steer_vectors.ReftIntervention

ReftIntervention(
    rotate_layer: Any,
    learned_source_weight: Any,
    learned_source_bias: Any = None,
    layer: int | None = None,
)

Bases: Payload

A LoReFT intervention: rotation basis + learned source map.

Accepted by the loreft algorithm. layer fixes the target layer (as recorded in a checkpoint); when None, VectorSpec.layers is required and the intervention is applied to each listed layer.

Source code in vllm-steer/vllm/model_hooks/steering/payloads.py

def __init__(
    self,
    rotate_layer: Any,
    learned_source_weight: Any,
    learned_source_bias: Any = None,
    layer: int | None = None,
):
    self.rotate_layer = _as_array("rotate_layer", rotate_layer, ndim=2)
    self.learned_source_weight = _as_array(
        "learned_source_weight", learned_source_weight, ndim=2
    )
    self.learned_source_bias = (
        None
        if learned_source_bias is None
        else _as_array("learned_source_bias", learned_source_bias, ndim=1)
    )
    hidden_size, rank = self.rotate_layer.shape
    if self.learned_source_weight.shape != (rank, hidden_size):
        raise ValueError(
            f"learned_source_weight shape {self.learned_source_weight.shape} "
            f"must be {(rank, hidden_size)} for rotate_layer shape "
            f"{self.rotate_layer.shape}"
        )
    if self.learned_source_bias is not None and self.learned_source_bias.shape != (
        rank,
    ):
        raise ValueError(
            f"learned_source_bias size {self.learned_source_bias.shape[0]} "
            f"!= rotation rank {rank}"
        )
    self.layer = None if layer is None else int(layer)

kind class-attribute instance-attribute

kind = 'reft'

rotate_layer instance-attribute

rotate_layer = _as_array(
    "rotate_layer", rotate_layer, ndim=2
)

learned_source_weight instance-attribute

learned_source_weight = _as_array(
    "learned_source_weight", learned_source_weight, ndim=2
)

learned_source_bias instance-attribute

learned_source_bias = (
    None
    if learned_source_bias is None
    else _as_array(
        "learned_source_bias", learned_source_bias, ndim=1
    )
)

layer instance-attribute

layer = None if layer is None else int(layer)

Concept substitution

h1 and h2 are DirectionVector objects or layer-to-vector dictionaries. They must have identical layer sets and matching widths at each layer. The returned ConceptPair preserves the two roles explicitly.

vllm.steer_vectors.ConceptPair

ConceptPair(
    h1: DirectionVector | dict, h2: DirectionVector | dict
)

Bases: Payload

Concept substitution directions: replace the h1 component with h2.

Accepted by the concept_replace algorithm. Both sides carry the same layer set; the roles are explicit — there is no filename or ordering heuristic.

Source code in vllm-steer/vllm/model_hooks/steering/payloads.py

def __init__(self, h1: "DirectionVector | dict", h2: "DirectionVector | dict"):
    self.h1 = h1 if isinstance(h1, DirectionVector) else DirectionVector(h1)
    self.h2 = h2 if isinstance(h2, DirectionVector) else DirectionVector(h2)
    if set(self.h1.layers) != set(self.h2.layers):
        raise ValueError(
            f"h1 layers {sorted(self.h1.layers)} != h2 layers "
            f"{sorted(self.h2.layers)}"
        )
    for layer in self.h1.layers:
        _check_dims(
            {"h1": self.h1.layers[layer], "h2": self.h2.layers[layer]},
            {"h1": 0, "h2": 0},
        )

kind class-attribute instance-attribute

kind = 'concept_pair'

h1 instance-attribute

h1 = (
    h1
    if isinstance(h1, DirectionVector)
    else DirectionVector(h1)
)

h2 instance-attribute

h2 = (
    h2
    if isinstance(h2, DirectionVector)
    else DirectionVector(h2)
)

MoE routing configuration

layers maps layer IDs to routing dictionaries. The returned RouterConfig normalizes and validates each dictionary:

Key	Meaning
mode	activate (default), deactivate, soft, soft_topk, or soft_random.
expert_ids	Nonnegative expert IDs affected by the selected mode.
activate_ids, deactivate_ids	Additional explicit lists accepted by hard activation/deactivation modes.
epsilon	Hard-routing margin, default 0.01.
lambda	Soft-routing strength, default 0.5.
topk	Positive expert count for soft_topk, default 8.

router = RouterConfig({10: {"mode": "deactivate", "expert_ids": [1, 3]}})

vllm.steer_vectors.RouterConfig

RouterConfig(layers: dict[int, dict[str, Any]])

Bases: Payload

Per-layer expert interventions, shared by JSON and in-memory inputs.

Source code in vllm-steer/vllm/model_hooks/steering/payloads.py

def __init__(self, layers: dict[int, dict[str, Any]]):
    if not layers:
        raise ValueError("RouterConfig requires at least one layer")
    self.layers: dict[int, dict[str, Any]] = {}
    for layer, config in layers.items():
        layer = int(layer)
        if layer < 0:
            raise ValueError("router layer ids must be non-negative")
        mode = config.get("mode", "activate")
        try:
            validate_router_mode(mode)
        except ValueError as exc:
            raise ValueError(f"Layer {layer}: {exc}") from None

        normalized = {
            "mode": mode,
            "expert_ids": _router_expert_ids(config, "expert_ids", layer),
        }
        if mode in ("activate", "deactivate"):
            normalized.update(
                activate_ids=_router_expert_ids(config, "activate_ids", layer),
                deactivate_ids=_router_expert_ids(config, "deactivate_ids", layer),
                epsilon=float(config.get("epsilon", 0.01)),
            )
            if not any(
                normalized[k]
                for k in ("expert_ids", "activate_ids", "deactivate_ids")
            ):
                raise ValueError(f"Layer {layer}: {mode} config has no expert ids")
            if not math.isfinite(normalized["epsilon"]):
                raise ValueError(f"Layer {layer}: epsilon must be finite")
        else:
            if not normalized["expert_ids"]:
                raise ValueError(f"Layer {layer}: {mode} config has no expert ids")
            normalized["lambda"] = float(config.get("lambda", 0.5))
            if not math.isfinite(normalized["lambda"]):
                raise ValueError(f"Layer {layer}: lambda must be finite")
            if mode == "soft_topk":
                topk = config.get("topk", 8)
                if not isinstance(topk, int) or isinstance(topk, bool) or topk < 1:
                    raise ValueError(
                        f"Layer {layer}: topk must be a positive integer"
                    )
                normalized["topk"] = topk
        if layer in self.layers:
            raise ValueError(f"duplicate router layer id: {layer}")
        self.layers[layer] = normalized

kind class-attribute instance-attribute

kind = 'router'

layers instance-attribute

layers: dict[int, dict[str, Any]] = {}

Pass router as VectorSpec.data with algorithm="moe_router". Supported graph modes and overrides are listed in the algorithm reference.