easysteer.steer

Analysis-based extraction of steering vectors from captured hidden states.

Unified extraction interface

easysteer.steer.extract_statistical_control_vector

extract_statistical_control_vector(
    method: str,
    all_hidden_states,
    positive_indices,
    negative_indices=None,
    **kwargs,
) -> StatisticalControlVector

Unified control vector extraction interface.

Parameters:

Name	Type	Description	Default
method	str	Method name; one of "diffmean", "pca", "lat", "linear_probe", "iti". ITI also requires validation captures, validation labels and query head counts.	required
all_hidden_states	list | CaptureResult	Nested [sample][layer][token] hidden states, or a CaptureResult from easysteer.hidden_states.	required
positive_indices	list[int]	Indices of positive samples.	required
negative_indices	list[int] | None	Indices of negative samples. If None, every sample index not in positive_indices becomes a negative, in ascending sample order.	None
**kwargs	Any	Method-specific options. Unknown options raise ValueError naming the accepted ones for method.	{}

Returns:

Name	Type	Description
StatisticalControlVector	StatisticalControlVector	The extracted control vector.

Source code in easysteer/steer/unified_interface.py

def extract_statistical_control_vector(
    method: str, all_hidden_states, positive_indices, negative_indices=None, **kwargs
) -> StatisticalControlVector:
    """Unified control vector extraction interface.

    Args:
        method (str): Method name; one of "diffmean", "pca", "lat",
            "linear_probe", "iti". ITI also requires validation captures,
            validation labels and query head counts.
        all_hidden_states (list | CaptureResult): Nested
            `[sample][layer][token]` hidden states, or a CaptureResult
            from easysteer.hidden_states.
        positive_indices (list[int]): Indices of positive samples.
        negative_indices (list[int] | None): Indices of negative
            samples. If None, every sample index not in
            ``positive_indices`` becomes a negative, in ascending
            sample order.
        **kwargs (Any): Method-specific options. Unknown options raise
            ValueError naming the accepted ones for ``method``.

    Returns:
        StatisticalControlVector: The extracted control vector.
    """
    return _dispatch(
        method, all_hidden_states, positive_indices, negative_indices, kwargs
    )

easysteer.steer.extract_diffmean_control_vector

extract_diffmean_control_vector(
    all_hidden_states,
    positive_indices,
    negative_indices=None,
    **kwargs,
) -> StatisticalControlVector

Extract a DiffMean control vector.

Parameters:

Name	Type	Description	Default
all_hidden_states	list | CaptureResult	Nested [sample][layer][token] hidden states, or a CaptureResult from easysteer.hidden_states.	required
positive_indices	list[int]	Indices of positive samples.	required
negative_indices	list[int] | None	Indices of negative samples. If None, every sample index not in positive_indices becomes a negative, in ascending sample order.	None
**kwargs	Any	Options accepted by DiffMeanExtractor: normalize (bool, default True), token_pos (int | str, default -1). Unknown options raise ValueError.	{}

Returns:

Name	Type	Description
StatisticalControlVector	StatisticalControlVector	The DiffMean control vector.

Source code in easysteer/steer/unified_interface.py

def extract_diffmean_control_vector(
    all_hidden_states, positive_indices, negative_indices=None, **kwargs
) -> StatisticalControlVector:
    """Extract a DiffMean control vector.

    Args:
        all_hidden_states (list | CaptureResult): Nested
            `[sample][layer][token]` hidden states, or a CaptureResult
            from easysteer.hidden_states.
        positive_indices (list[int]): Indices of positive samples.
        negative_indices (list[int] | None): Indices of negative
            samples. If None, every sample index not in
            ``positive_indices`` becomes a negative, in ascending
            sample order.
        **kwargs (Any): Options accepted by DiffMeanExtractor:
            `normalize` (bool, default True),
            `token_pos` (int | str, default -1). Unknown options raise
            ValueError.

    Returns:
        StatisticalControlVector: The DiffMean control vector.
    """
    return _dispatch(
        "diffmean", all_hidden_states, positive_indices, negative_indices, kwargs
    )

easysteer.steer.extract_pca_control_vector

extract_pca_control_vector(
    all_hidden_states,
    positive_indices,
    negative_indices=None,
    **kwargs,
) -> StatisticalControlVector

Extract a PCA control vector.

Parameters:

Name	Type	Description	Default
all_hidden_states	list | CaptureResult	Nested [sample][layer][token] hidden states, or a CaptureResult from easysteer.hidden_states.	required
positive_indices	list[int]	Indices of positive samples.	required
negative_indices	list[int] | None	Indices of negative samples. If None, every sample index not in positive_indices becomes a negative, in ascending sample order.	None
**kwargs	Any	Options accepted by PCAExtractor: method (str, default "standard") selects the PCA variant ("standard" uses only positive samples, "diff" runs PCA over positive/negative differences, "center" over pair-centered samples); correct_direction (bool, default True) flips the vector if needed so it points from negative toward positive samples; n_components (int, must be 1); normalize (bool, default True); token_pos (int | str, default -1, the last token). Unknown options raise ValueError.	{}

Returns:

Name	Type	Description
StatisticalControlVector	StatisticalControlVector	The PCA control vector.

Examples:

>>> # Plain PCA over positive samples only
>>> pca_vector = extract_pca_control_vector(
...     all_hidden_states, positive_indices,
...     method="standard"
... )
>>>
>>> # PCA over pair differences with direction correction
>>> pca_diff_vector = extract_pca_control_vector(
...     all_hidden_states, positive_indices, negative_indices,
...     method="diff", correct_direction=True
... )
>>>
>>> # PCA over pair differences without direction correction
>>> pca_diff_no_correct = extract_pca_control_vector(
...     all_hidden_states, positive_indices, negative_indices,
...     method="diff", correct_direction=False
... )

Source code in easysteer/steer/unified_interface.py

def extract_pca_control_vector(
    all_hidden_states, positive_indices, negative_indices=None, **kwargs
) -> StatisticalControlVector:
    """Extract a PCA control vector.

    Args:
        all_hidden_states (list | CaptureResult): Nested
            `[sample][layer][token]` hidden states, or a CaptureResult
            from easysteer.hidden_states.
        positive_indices (list[int]): Indices of positive samples.
        negative_indices (list[int] | None): Indices of negative
            samples. If None, every sample index not in
            ``positive_indices`` becomes a negative, in ascending
            sample order.
        **kwargs (Any): Options accepted by PCAExtractor:
            `method` (str, default "standard") selects the PCA variant
            ("standard" uses only positive samples, "diff" runs PCA
            over positive/negative differences, "center" over
            pair-centered samples); `correct_direction` (bool, default
            True) flips the vector if needed so it points from negative
            toward positive samples; `n_components` (int, must be 1);
            `normalize` (bool, default True); `token_pos` (int | str,
            default -1, the last token). Unknown
            options raise ValueError.

    Returns:
        StatisticalControlVector: The PCA control vector.

    Examples:
        >>> # Plain PCA over positive samples only
        >>> pca_vector = extract_pca_control_vector(
        ...     all_hidden_states, positive_indices,
        ...     method="standard"
        ... )
        >>>
        >>> # PCA over pair differences with direction correction
        >>> pca_diff_vector = extract_pca_control_vector(
        ...     all_hidden_states, positive_indices, negative_indices,
        ...     method="diff", correct_direction=True
        ... )
        >>>
        >>> # PCA over pair differences without direction correction
        >>> pca_diff_no_correct = extract_pca_control_vector(
        ...     all_hidden_states, positive_indices, negative_indices,
        ...     method="diff", correct_direction=False
        ... )
    """
    return _dispatch(
        "pca", all_hidden_states, positive_indices, negative_indices, kwargs
    )

easysteer.steer.extract_lat_control_vector

extract_lat_control_vector(
    all_hidden_states,
    positive_indices,
    negative_indices=None,
    **kwargs,
) -> StatisticalControlVector

Extract a LAT control vector.

Parameters:

Name	Type	Description	Default
all_hidden_states	list | CaptureResult	Nested [sample][layer][token] hidden states, or a CaptureResult from easysteer.hidden_states.	required
positive_indices	list[int]	Indices of positive samples.	required
negative_indices	list[int] | None	Indices of negative samples. If None and use_positive_only is False, every sample index not in positive_indices becomes a negative, in ascending sample order.	None
**kwargs	Any	Options accepted by LATExtractor: use_positive_only (bool, default True) uses only the positive samples; correct_direction (bool, default True) flips the vector if needed so it points from negative toward positive samples; n_components (int, default 1); normalize (bool, default True); token_pos (int | str, default -1, the last token). Unknown options raise ValueError.	{}

Returns:

Name	Type	Description
StatisticalControlVector	StatisticalControlVector	The LAT control vector.

Examples:

>>> # Positive samples only (traditional LAT)
>>> lat_vector = extract_lat_control_vector(
...     all_hidden_states, positive_indices,
...     use_positive_only=True
... )
>>>
>>> # Positive and negative samples with direction correction
>>> lat_mixed_vector = extract_lat_control_vector(
...     all_hidden_states, positive_indices, negative_indices,
...     use_positive_only=False, correct_direction=True
... )

Source code in easysteer/steer/unified_interface.py

def extract_lat_control_vector(
    all_hidden_states, positive_indices, negative_indices=None, **kwargs
) -> StatisticalControlVector:
    """Extract a LAT control vector.

    Args:
        all_hidden_states (list | CaptureResult): Nested
            `[sample][layer][token]` hidden states, or a CaptureResult
            from easysteer.hidden_states.
        positive_indices (list[int]): Indices of positive samples.
        negative_indices (list[int] | None): Indices of negative
            samples. If None and `use_positive_only` is False, every
            sample index not in ``positive_indices`` becomes a
            negative, in ascending sample order.
        **kwargs (Any): Options accepted by LATExtractor:
            `use_positive_only` (bool, default True) uses only the
            positive samples; `correct_direction` (bool, default True)
            flips the vector if needed so it points from negative
            toward positive samples; `n_components` (int, default 1);
            `normalize` (bool, default True); `token_pos` (int | str,
            default -1, the last token). Unknown
            options raise ValueError.

    Returns:
        StatisticalControlVector: The LAT control vector.

    Examples:
        >>> # Positive samples only (traditional LAT)
        >>> lat_vector = extract_lat_control_vector(
        ...     all_hidden_states, positive_indices,
        ...     use_positive_only=True
        ... )
        >>>
        >>> # Positive and negative samples with direction correction
        >>> lat_mixed_vector = extract_lat_control_vector(
        ...     all_hidden_states, positive_indices, negative_indices,
        ...     use_positive_only=False, correct_direction=True
        ... )
    """
    return _dispatch(
        "lat", all_hidden_states, positive_indices, negative_indices, kwargs
    )

easysteer.steer.extract_linear_probe_control_vector

extract_linear_probe_control_vector(
    all_hidden_states,
    positive_indices,
    negative_indices=None,
    **kwargs,
) -> StatisticalControlVector

Extract a Linear Probe control vector.

Parameters:

Name	Type	Description	Default
all_hidden_states	list | CaptureResult	Nested [sample][layer][token] hidden states, or a CaptureResult from easysteer.hidden_states.	required
positive_indices	list[int]	Indices of positive samples.	required
negative_indices	list[int] | None	Indices of negative samples. If None, every sample index not in positive_indices becomes a negative, in ascending sample order.	None
**kwargs	Any	Options accepted by LinearProbeExtractor: regularization (str, default "l2") is one of "l1", "l2", "elasticnet", "none" — unknown values raise ValueError; C (float, default 1.0) is the inverse regularization strength (for L2, C=1.0 is usually fine; for L1, prefer C=10.0 or larger to avoid excessive sparsification; with "none" C is ignored); standardize (bool, default True); normalize (bool, default True); token_pos (int | str, default -1, the last token). Unknown options raise ValueError.	{}

Returns:

Name	Type	Description
StatisticalControlVector	StatisticalControlVector	The Linear Probe control vector.

Examples:

>>> # L2 regularization (recommended)
>>> linear_probe_vector = extract_linear_probe_control_vector(
...     all_hidden_states, positive_indices, negative_indices,
...     regularization="l2", C=1.0
... )
>>>
>>> # L1 regularization (feature selection)
>>> linear_probe_l1 = extract_linear_probe_control_vector(
...     all_hidden_states, positive_indices, negative_indices,
...     regularization="l1", C=10.0
... )

Source code in easysteer/steer/unified_interface.py

def extract_linear_probe_control_vector(
    all_hidden_states, positive_indices, negative_indices=None, **kwargs
) -> StatisticalControlVector:
    """Extract a Linear Probe control vector.

    Args:
        all_hidden_states (list | CaptureResult): Nested
            `[sample][layer][token]` hidden states, or a CaptureResult
            from easysteer.hidden_states.
        positive_indices (list[int]): Indices of positive samples.
        negative_indices (list[int] | None): Indices of negative
            samples. If None, every sample index not in
            ``positive_indices`` becomes a negative, in ascending
            sample order.
        **kwargs (Any): Options accepted by LinearProbeExtractor:
            `regularization` (str, default "l2") is one of "l1", "l2",
            "elasticnet", "none" — unknown values raise ValueError;
            `C` (float, default 1.0) is the inverse regularization
            strength (for L2, C=1.0 is usually fine; for L1, prefer
            C=10.0 or larger to avoid excessive sparsification; with
            "none" C is ignored); `standardize` (bool, default True);
            `normalize` (bool, default True); `token_pos` (int | str,
            default -1, the last token). Unknown
            options raise ValueError.

    Returns:
        StatisticalControlVector: The Linear Probe control vector.

    Examples:
        >>> # L2 regularization (recommended)
        >>> linear_probe_vector = extract_linear_probe_control_vector(
        ...     all_hidden_states, positive_indices, negative_indices,
        ...     regularization="l2", C=1.0
        ... )
        >>>
        >>> # L1 regularization (feature selection)
        >>> linear_probe_l1 = extract_linear_probe_control_vector(
        ...     all_hidden_states, positive_indices, negative_indices,
        ...     regularization="l1", C=10.0
        ... )
    """
    return _dispatch(
        "linear_probe", all_hidden_states, positive_indices, negative_indices, kwargs
    )

ITI attention extraction

easysteer.steer.ITIExtractor

Select heads with probes and build mean-difference ITI directions.

Probes fit training rows and rank heads by validation accuracy. Following ITI, the final direction and its projection standard deviation use both development splits. Split by question before capture; no held-out test answers should appear in either input.

Source code in easysteer/steer/iti.py

class ITIExtractor:
    """Select heads with probes and build mean-difference ITI directions.

    Probes fit training rows and rank heads by validation accuracy. Following
    ITI, the final direction and its projection standard deviation use both
    development splits. Split by question before capture; no held-out test
    answers should appear in either input.
    """

    @staticmethod
    def extract(
        all_hidden_states,
        positive_indices,
        negative_indices=None,
        *,
        validation_hidden_states,
        validation_positive_indices,
        validation_negative_indices=None,
        num_heads: dict[int, int],
        top_k: int = 1,
        token_pos: int | str = -1,
    ) -> StatisticalControlVector:
        """Return concatenated per-layer ``sigma * direction`` vectors.

        Inputs use the other extractors' CaptureResult or nested-list layout.
        ``num_heads`` maps layer ids to query head counts; for capture input,
        use ``{layer: layout['num_heads'] for layer, layout in result.layouts.items()}``.
        Each row must contain contiguous head outputs. Unselected head slices
        remain zero; layers with no selected head are omitted. Apply with
        ``attention_add`` and no normalization, using ``scale`` for ITI's alpha.
        """
        train, train_labels = _labelled_rows(
            all_hidden_states, positive_indices, negative_indices, token_pos
        )
        validation, validation_labels = _labelled_rows(
            validation_hidden_states,
            validation_positive_indices,
            validation_negative_indices,
            token_pos,
        )
        if not train or set(train) != set(validation) or set(train) != set(num_heads):
            raise ValueError(
                "training, validation and num_heads must have the same layers"
            )
        if any(type(count) is not int or count < 1 for count in num_heads.values()):
            raise ValueError("num_heads must contain positive query head counts")
        if type(top_k) is not int or not 1 <= top_k <= sum(num_heads.values()):
            raise ValueError("top_k must be between 1 and the total query head count")

        candidates = []
        for layer in sorted(train):
            width = train[layer].shape[1]
            count = num_heads[layer]
            if validation[layer].shape[1] != width or width == 0 or width % count:
                raise ValueError(f"layer {layer}: incompatible attention head widths")
            head_size = width // count
            for head in range(count):
                head_slice = slice(head * head_size, (head + 1) * head_size)
                probe = LogisticRegression(max_iter=1000, random_state=42).fit(
                    train[layer][:, head_slice], train_labels
                )
                score = float(
                    probe.score(validation[layer][:, head_slice], validation_labels)
                )
                candidates.append((score, layer, head, head_slice))

        directions, scores, scales = {}, {}, {}
        labels = np.concatenate([train_labels, validation_labels])
        # A fixed layer/head ordering resolves validation-score ties.
        candidates.sort(key=lambda item: (-item[0], item[1], item[2]))
        for score, layer, head, head_slice in candidates[:top_k]:
            rows = np.vstack(
                [train[layer][:, head_slice], validation[layer][:, head_slice]]
            )
            direction = l2_normalize(
                rows[labels == 1].mean(0) - rows[labels == 0].mean(0)
            )
            sigma = float(np.std(rows @ direction))
            if not np.isfinite(sigma) or sigma == 0:
                raise ValueError(
                    f"layer {layer}, head {head}: no nonzero finite ITI direction"
                )
            vector = directions.setdefault(
                layer, np.zeros(train[layer].shape[1], dtype=np.float32)
            )
            vector[head_slice] = sigma * direction
            key = f"{layer}.{head}"
            scores[key], scales[key] = score, sigma

        return StatisticalControlVector(
            method="iti",
            directions=directions,
            metadata=_metadata(
                normalize=False,
                n_positive=int(np.sum(train_labels)),
                n_negative=int(np.sum(1 - train_labels)),
                n_validation_positive=int(np.sum(validation_labels)),
                n_validation_negative=int(np.sum(1 - validation_labels)),
                token_pos=token_pos,
                top_k=top_k,
                num_heads=num_heads,
                selected_head_scores=scores,
                projection_std=scales,
                direction_data="train+validation",
            ),
        )

extract staticmethod

extract(
    all_hidden_states,
    positive_indices,
    negative_indices=None,
    *,
    validation_hidden_states,
    validation_positive_indices,
    validation_negative_indices=None,
    num_heads: dict[int, int],
    top_k: int = 1,
    token_pos: int | str = -1,
) -> StatisticalControlVector

Return concatenated per-layer sigma * direction vectors.

Inputs use the other extractors' CaptureResult or nested-list layout. num_heads maps layer ids to query head counts; for capture input, use {layer: layout['num_heads'] for layer, layout in result.layouts.items()}. Each row must contain contiguous head outputs. Unselected head slices remain zero; layers with no selected head are omitted. Apply with attention_add and no normalization, using scale for ITI's alpha.

Source code in easysteer/steer/iti.py

@staticmethod
def extract(
    all_hidden_states,
    positive_indices,
    negative_indices=None,
    *,
    validation_hidden_states,
    validation_positive_indices,
    validation_negative_indices=None,
    num_heads: dict[int, int],
    top_k: int = 1,
    token_pos: int | str = -1,
) -> StatisticalControlVector:
    """Return concatenated per-layer ``sigma * direction`` vectors.

    Inputs use the other extractors' CaptureResult or nested-list layout.
    ``num_heads`` maps layer ids to query head counts; for capture input,
    use ``{layer: layout['num_heads'] for layer, layout in result.layouts.items()}``.
    Each row must contain contiguous head outputs. Unselected head slices
    remain zero; layers with no selected head are omitted. Apply with
    ``attention_add`` and no normalization, using ``scale`` for ITI's alpha.
    """
    train, train_labels = _labelled_rows(
        all_hidden_states, positive_indices, negative_indices, token_pos
    )
    validation, validation_labels = _labelled_rows(
        validation_hidden_states,
        validation_positive_indices,
        validation_negative_indices,
        token_pos,
    )
    if not train or set(train) != set(validation) or set(train) != set(num_heads):
        raise ValueError(
            "training, validation and num_heads must have the same layers"
        )
    if any(type(count) is not int or count < 1 for count in num_heads.values()):
        raise ValueError("num_heads must contain positive query head counts")
    if type(top_k) is not int or not 1 <= top_k <= sum(num_heads.values()):
        raise ValueError("top_k must be between 1 and the total query head count")

    candidates = []
    for layer in sorted(train):
        width = train[layer].shape[1]
        count = num_heads[layer]
        if validation[layer].shape[1] != width or width == 0 or width % count:
            raise ValueError(f"layer {layer}: incompatible attention head widths")
        head_size = width // count
        for head in range(count):
            head_slice = slice(head * head_size, (head + 1) * head_size)
            probe = LogisticRegression(max_iter=1000, random_state=42).fit(
                train[layer][:, head_slice], train_labels
            )
            score = float(
                probe.score(validation[layer][:, head_slice], validation_labels)
            )
            candidates.append((score, layer, head, head_slice))

    directions, scores, scales = {}, {}, {}
    labels = np.concatenate([train_labels, validation_labels])
    # A fixed layer/head ordering resolves validation-score ties.
    candidates.sort(key=lambda item: (-item[0], item[1], item[2]))
    for score, layer, head, head_slice in candidates[:top_k]:
        rows = np.vstack(
            [train[layer][:, head_slice], validation[layer][:, head_slice]]
        )
        direction = l2_normalize(
            rows[labels == 1].mean(0) - rows[labels == 0].mean(0)
        )
        sigma = float(np.std(rows @ direction))
        if not np.isfinite(sigma) or sigma == 0:
            raise ValueError(
                f"layer {layer}, head {head}: no nonzero finite ITI direction"
            )
        vector = directions.setdefault(
            layer, np.zeros(train[layer].shape[1], dtype=np.float32)
        )
        vector[head_slice] = sigma * direction
        key = f"{layer}.{head}"
        scores[key], scales[key] = score, sigma

    return StatisticalControlVector(
        method="iti",
        directions=directions,
        metadata=_metadata(
            normalize=False,
            n_positive=int(np.sum(train_labels)),
            n_negative=int(np.sum(1 - train_labels)),
            n_validation_positive=int(np.sum(validation_labels)),
            n_validation_negative=int(np.sum(1 - validation_labels)),
            token_pos=token_pos,
            top_k=top_k,
            num_heads=num_heads,
            selected_head_scores=scores,
            projection_std=scales,
            direction_data="train+validation",
        ),
    )

Containers and utilities

easysteer.steer.StatisticalControlVector dataclass

Statistical control vector with per-layer directions.

Source code in easysteer/steer/utils.py

@dataclasses.dataclass
class StatisticalControlVector:
    """Statistical control vector with per-layer directions."""

    method: str
    directions: dict[int, np.ndarray]
    metadata: dict = None
    # Only echoed into the gguf "model_hint" field for repeng
    # compatibility; nothing in EasySteer consumes it.
    model_type: str = "unknown"

    def export_gguf(self, path: os.PathLike[str] | str):
        """Export directions and metadata to a repeng-compatible GGUF file."""
        import gguf

        arch = "controlvector"
        writer = gguf.GGUFWriter(path, arch)
        writer.add_string(f"{arch}.model_hint", self.model_type)
        writer.add_string(f"{arch}.method", self.method)
        writer.add_uint32(f"{arch}.layer_count", len(self.directions))

        if self.metadata:
            for key, value in self.metadata.items():
                if isinstance(value, (int, float)):
                    writer.add_float32(f"{arch}.{key}", float(value))
                elif isinstance(value, str):
                    writer.add_string(f"{arch}.{key}", value)
                elif isinstance(value, dict):
                    for subkey, subvalue in value.items():
                        if isinstance(subvalue, (int, float)):
                            writer.add_float32(
                                f"{arch}.{key}.{subkey}", float(subvalue)
                            )

        for layer in self.directions.keys():
            writer.add_tensor(f"direction.{layer}", self.directions[layer])

        writer.write_header_to_file()
        writer.write_kv_data_to_file()
        writer.write_tensors_to_file()
        writer.close()

    @classmethod
    def import_gguf(cls, path: os.PathLike[str] | str) -> "StatisticalControlVector":
        """Import a StatisticalControlVector from a GGUF file."""
        import gguf

        reader = gguf.GGUFReader(path)

        archf = reader.get_field("general.architecture")
        if not archf or not len(archf.parts):
            warnings.warn(".gguf file missing architecture field")
        else:
            arch = str(bytes(archf.parts[-1]), encoding="utf-8", errors="replace")
            if arch != "controlvector":
                warnings.warn(
                    f".gguf file with architecture {arch!r} does not "
                    f"appear to be a control vector!"
                )

        modelf = reader.get_field("controlvector.model_hint")
        if not modelf or not len(modelf.parts):
            raise ValueError(".gguf file missing controlvector.model_hint field")
        model_hint = str(bytes(modelf.parts[-1]), encoding="utf-8")

        methodf = reader.get_field("controlvector.method")
        method = "unknown"
        if methodf and len(methodf.parts):
            method = str(bytes(methodf.parts[-1]), encoding="utf-8")

        directions = {}
        metadata = {}

        skipped_suffixes = (".model_hint", ".method", ".layer_count")
        for field_name, field in reader.fields.items():
            if field_name.startswith("controlvector.") and not (
                field_name.endswith(skipped_suffixes)
            ):
                key = field_name.replace("controlvector.", "")
                if field.types == [gguf.GGMLQuantizationType.F32]:
                    metadata[key] = float(field.parts[0])
                elif field.types == [gguf.GGMLQuantizationType.I32]:
                    metadata[key] = int(field.parts[0])

        for tensor in reader.tensors:
            if not tensor.name.startswith("direction."):
                continue
            try:
                layer = int(tensor.name.split(".")[1])
            except (IndexError, ValueError):
                raise ValueError(
                    f".gguf file has invalid direction field name: {tensor.name}"
                )
            directions[layer] = tensor.data

        return cls(
            model_type=model_hint,
            method=method,
            directions=directions,
            metadata=metadata,
        )

export_gguf

export_gguf(path: PathLike[str] | str)

Export directions and metadata to a repeng-compatible GGUF file.

Source code in easysteer/steer/utils.py

def export_gguf(self, path: os.PathLike[str] | str):
    """Export directions and metadata to a repeng-compatible GGUF file."""
    import gguf

    arch = "controlvector"
    writer = gguf.GGUFWriter(path, arch)
    writer.add_string(f"{arch}.model_hint", self.model_type)
    writer.add_string(f"{arch}.method", self.method)
    writer.add_uint32(f"{arch}.layer_count", len(self.directions))

    if self.metadata:
        for key, value in self.metadata.items():
            if isinstance(value, (int, float)):
                writer.add_float32(f"{arch}.{key}", float(value))
            elif isinstance(value, str):
                writer.add_string(f"{arch}.{key}", value)
            elif isinstance(value, dict):
                for subkey, subvalue in value.items():
                    if isinstance(subvalue, (int, float)):
                        writer.add_float32(
                            f"{arch}.{key}.{subkey}", float(subvalue)
                        )

    for layer in self.directions.keys():
        writer.add_tensor(f"direction.{layer}", self.directions[layer])

    writer.write_header_to_file()
    writer.write_kv_data_to_file()
    writer.write_tensors_to_file()
    writer.close()

import_gguf classmethod

import_gguf(
    path: PathLike[str] | str,
) -> StatisticalControlVector

Import a StatisticalControlVector from a GGUF file.

Source code in easysteer/steer/utils.py

@classmethod
def import_gguf(cls, path: os.PathLike[str] | str) -> "StatisticalControlVector":
    """Import a StatisticalControlVector from a GGUF file."""
    import gguf

    reader = gguf.GGUFReader(path)

    archf = reader.get_field("general.architecture")
    if not archf or not len(archf.parts):
        warnings.warn(".gguf file missing architecture field")
    else:
        arch = str(bytes(archf.parts[-1]), encoding="utf-8", errors="replace")
        if arch != "controlvector":
            warnings.warn(
                f".gguf file with architecture {arch!r} does not "
                f"appear to be a control vector!"
            )

    modelf = reader.get_field("controlvector.model_hint")
    if not modelf or not len(modelf.parts):
        raise ValueError(".gguf file missing controlvector.model_hint field")
    model_hint = str(bytes(modelf.parts[-1]), encoding="utf-8")

    methodf = reader.get_field("controlvector.method")
    method = "unknown"
    if methodf and len(methodf.parts):
        method = str(bytes(methodf.parts[-1]), encoding="utf-8")

    directions = {}
    metadata = {}

    skipped_suffixes = (".model_hint", ".method", ".layer_count")
    for field_name, field in reader.fields.items():
        if field_name.startswith("controlvector.") and not (
            field_name.endswith(skipped_suffixes)
        ):
            key = field_name.replace("controlvector.", "")
            if field.types == [gguf.GGMLQuantizationType.F32]:
                metadata[key] = float(field.parts[0])
            elif field.types == [gguf.GGMLQuantizationType.I32]:
                metadata[key] = int(field.parts[0])

    for tensor in reader.tensors:
        if not tensor.name.startswith("direction."):
            continue
        try:
            layer = int(tensor.name.split(".")[1])
        except (IndexError, ValueError):
            raise ValueError(
                f".gguf file has invalid direction field name: {tensor.name}"
            )
        directions[layer] = tensor.data

    return cls(
        model_type=model_hint,
        method=method,
        directions=directions,
        metadata=metadata,
    )

easysteer.steer.extract_token_hiddens

extract_token_hiddens(
    all_hidden_states,
    positive_indices,
    negative_indices=None,
    token_pos=-1,
) -> tuple[dict, dict]

Extract hidden states of one token position per sample.

Parameters:

Name	Type	Description	Default
all_hidden_states	list | CaptureResult	Nested [sample][layer][token] hidden states, where each entry is a tensor or numpy array, or a CaptureResult from easysteer.hidden_states.	required
positive_indices	list[int]	Indices of positive samples.	required
negative_indices	list[int] | None	Indices of negative samples. If None, every sample index not in positive_indices becomes a negative, in ascending sample order (the convention shared by all extractors).	None
token_pos	int | str	Token position to extract: an int index (-1 selects the last token, the default), "first", "last", "mean" (average over tokens), "max" or "min" (token with the largest/smallest L2 norm).	-1

Returns:

Type	Description
tuple[dict, dict]	tuple[dict, dict]: (positive_hiddens, negative_hiddens), each a dict mapping layer key to a (n_samples, hidden_dim) array. Layer keys are the model layer ids for CaptureResult input and positional indices for nested-list input.

Source code in easysteer/steer/utils.py

def extract_token_hiddens(
    all_hidden_states, positive_indices, negative_indices=None, token_pos=-1
) -> tuple[dict, dict]:
    """Extract hidden states of one token position per sample.

    Args:
        all_hidden_states (list | CaptureResult): Nested
            `[sample][layer][token]` hidden states, where each entry is
            a tensor or numpy array, or a CaptureResult from
            easysteer.hidden_states.
        positive_indices (list[int]): Indices of positive samples.
        negative_indices (list[int] | None): Indices of negative
            samples. If None, every sample index not in
            ``positive_indices`` becomes a negative, in ascending
            sample order (the convention shared by all extractors).
        token_pos (int | str): Token position to extract:
            an int index (-1 selects the last token, the default),
            "first", "last", "mean" (average over tokens), "max" or
            "min" (token with the largest/smallest L2 norm).

    Returns:
        tuple[dict, dict]: `(positive_hiddens, negative_hiddens)`, each
            a dict mapping layer key to a `(n_samples, hidden_dim)`
            array. Layer keys are the model layer ids for CaptureResult
            input and positional indices for nested-list input.
    """
    layer_keys = None
    if hasattr(all_hidden_states, "to_nested"):
        layer_keys = list(all_hidden_states.layer_ids)
        all_hidden_states = all_hidden_states.to_nested()
    if negative_indices is None:
        negative_indices = derive_negative_indices(
            len(all_hidden_states), positive_indices
        )

    n_layers = len(all_hidden_states[0])
    if layer_keys is None:
        layer_keys = list(range(n_layers))

    def collect(indices):
        """Gather one token row per sample for each layer key."""
        hiddens = {layer: [] for layer in layer_keys}
        for sample_idx in indices:
            sample_hiddens = all_hidden_states[sample_idx]
            for layer_pos, layer_key in enumerate(layer_keys):
                token_hidden = extract_token_from_sequence(
                    sample_hiddens[layer_pos], token_pos
                )
                hiddens[layer_key].append(_to_numpy(token_hidden))
        return hiddens

    positive_hiddens = collect(positive_indices)
    negative_hiddens = collect(negative_indices or [])

    positive_hiddens = {k: np.vstack(v) for k, v in positive_hiddens.items()}
    if negative_indices and any(negative_hiddens.values()):
        negative_hiddens = {k: np.vstack(v) for k, v in negative_hiddens.items()}
    else:
        negative_hiddens = {}

    return positive_hiddens, negative_hiddens

SAE helpers

easysteer.steer.search_sae_features

search_sae_features(
    model_id: str,
    sae_id: str,
    query: str,
    api_key: str | None = None,
) -> list[dict[str, Any]]

Call SAEFeatureExplorer.search_features without retaining a client.

api_key defaults to the NP_API_KEY environment variable.

Source code in easysteer/steer/sae.py

def search_sae_features(
    model_id: str, sae_id: str, query: str, api_key: str | None = None
) -> list[dict[str, Any]]:
    """Call SAEFeatureExplorer.search_features without retaining a client.

    api_key defaults to the NP_API_KEY environment variable.
    """
    explorer = SAEFeatureExplorer(api_key=api_key)
    return explorer.search_features(model_id, sae_id, query)

easysteer.steer.get_sae_feature_explanation

get_sae_feature_explanation(
    model_id: str,
    sae_id: str,
    feature_index: int,
    api_key: str | None = None,
) -> dict[str, Any]

Call SAEFeatureExplorer.get_feature_explanation with a temporary client.

api_key defaults to the NP_API_KEY environment variable.

Source code in easysteer/steer/sae.py

def get_sae_feature_explanation(
    model_id: str, sae_id: str, feature_index: int, api_key: str | None = None
) -> dict[str, Any]:
    """Call SAEFeatureExplorer.get_feature_explanation with a temporary client.

    api_key defaults to the NP_API_KEY environment variable.
    """
    explorer = SAEFeatureExplorer(api_key=api_key)
    return explorer.get_feature_explanation(model_id, sae_id, feature_index)

easysteer.steer.extract_sae_decoder_vector

extract_sae_decoder_vector(
    model_file: str,
    feature_index: int,
    save_path: str | None = None,
) -> np.ndarray | None

Call SAEFeatureExplorer.extract_decoder_vector and return its array.

Source code in easysteer/steer/sae.py

def extract_sae_decoder_vector(
    model_file: str, feature_index: int, save_path: str | None = None
) -> np.ndarray | None:
    """Call SAEFeatureExplorer.extract_decoder_vector and return its array."""
    explorer = SAEFeatureExplorer()
    return explorer.extract_decoder_vector(model_file, feature_index, save_path)

Payload adapters (easysteer.vectors)

Client-side adapters from third-party checkpoint formats to the canonical steering payloads passed via VectorSpec(data=...).

easysteer.vectors.load

load(path: str, *, format: str, **options) -> Payload

Load an explicitly identified checkpoint schema into a canonical payload.

Formats: gguf, concept_pair (named h1/h2 GGUF directory), moe_router (JSON), pt_direction, pyreft, lm_steer and linear_transport. Options go to the chosen adapter; for example layers=[10] for pt_direction or vector_index=1 for lm_steer. A suffix such as .pt never selects or guesses a checkpoint schema.

Source code in easysteer/vectors.py

def load(path: str, *, format: str, **options) -> Payload:
    """Load an explicitly identified checkpoint schema into a canonical payload.

    Formats: ``gguf``, ``concept_pair`` (named h1/h2 GGUF directory),
    ``moe_router`` (JSON), ``pt_direction``, ``pyreft``, ``lm_steer`` and
    ``linear_transport``. Options go to the chosen adapter; for example
    ``layers=[10]`` for pt_direction or ``vector_index=1`` for lm_steer.
    A suffix such as .pt never selects or guesses a checkpoint schema.
    """
    if format in ("gguf", "concept_pair", "moe_router"):
        from vllm.model_hooks.steering.loading import load_file_payload

        return load_file_payload(path, format=format, **options)
    adapters = {
        "pt_direction": from_pt_direction,
        "pyreft": from_pyreft,
        "lm_steer": from_lm_steer,
        "linear_transport": from_linear_transport,
    }
    if format not in adapters:
        raise ValueError(f"Unknown checkpoint format: {format!r}")
    return adapters[format](path, **options)

easysteer.vectors.to_json_payload

to_json_payload(payload: Payload) -> dict[str, Any]

Encode a canonical payload for JSON files or OpenAI HTTP requests.

Tensor bytes become base64 strings; shapes, metadata and the content hash remain identical to payload.to_wire(). This runs in the producing EasySteer environment, so JSON consumers need no engine or tensor dependencies.

Source code in easysteer/vectors.py

def to_json_payload(payload: Payload) -> dict[str, Any]:
    """Encode a canonical payload for JSON files or OpenAI HTTP requests.

    Tensor bytes become base64 strings; shapes, metadata and the content
    hash remain identical to ``payload.to_wire()``. This runs in the
    producing EasySteer environment, so JSON consumers need no engine
    or tensor dependencies.
    """
    wire = payload.to_wire()
    return {
        **wire,
        "tensors": {
            name: {**tensor, "data": base64.b64encode(tensor["data"]).decode("ascii")}
            for name, tensor in wire["tensors"].items()
        },
    }

easysteer.vectors.from_control_vector

from_control_vector(cv: Any) -> DirectionVector

Convert a StatisticalControlVector directly, without a GGUF round-trip.

Source code in easysteer/vectors.py

def from_control_vector(cv: Any) -> DirectionVector:
    """Convert a StatisticalControlVector directly, without a GGUF round-trip."""
    if not getattr(cv, "directions", None):
        raise ValueError("control vector has no directions")
    return DirectionVector(dict(cv.directions))

easysteer.vectors.from_gguf

from_gguf(path: str) -> DirectionVector

Payload from an EasySteer GGUF export (direction.<layer>).

Source code in easysteer/vectors.py

def from_gguf(path: str) -> DirectionVector:
    """Payload from an EasySteer GGUF export (``direction.<layer>``)."""
    from vllm.model_hooks.steering.loading import load_file_payload

    return load_file_payload(path, format="gguf")

easysteer.vectors.from_pt_direction

from_pt_direction(
    path: str, layers: list[int]
) -> DirectionVector

Payload from a bare direction tensor saved with torch.save.

The file holds one vector (tensor or numpy array); it is applied to each listed layer.

Source code in easysteer/vectors.py

def from_pt_direction(path: str, layers: list[int]) -> DirectionVector:
    """Payload from a bare direction tensor saved with ``torch.save``.

    The file holds one vector (tensor or numpy array); it is applied to
    each listed layer.
    """
    import numpy as np
    import torch

    if not layers:
        raise ValueError("layers must be non-empty")
    vector = torch.load(path, map_location="cpu", weights_only=False)
    if isinstance(vector, np.ndarray):
        vector = torch.from_numpy(vector)
    if not isinstance(vector, torch.Tensor):
        raise ValueError(
            f"{path} does not contain a tensor or numpy array: {type(vector).__name__}"
        )
    return DirectionVector({layer: vector for layer in layers})

easysteer.vectors.from_pyreft

from_pyreft(
    path: str,
) -> DirectionVector | ReftIntervention

Payload from a pyreft checkpoint directory.

Reads the single *.bin + config pair. A BiasIntervention-style state dict (one vector) becomes a DirectionVector for the direct algorithm; a LoReFT state dict (rotation + learned source) becomes a ReftIntervention for loreft. The checkpoint's layer index is preserved.

Source code in easysteer/vectors.py

def from_pyreft(path: str) -> DirectionVector | ReftIntervention:
    """Payload from a pyreft checkpoint directory.

    Reads the single ``*.bin`` + config pair. A BiasIntervention-style
    state dict (one vector) becomes a ``DirectionVector`` for the
    ``direct`` algorithm; a LoReFT state dict (rotation + learned
    source) becomes a ``ReftIntervention`` for ``loreft``. The
    checkpoint's layer index is preserved.
    """
    import torch

    bin_path, layer = _find_pyreft_checkpoint(path)
    state = torch.load(bin_path, map_location="cpu", weights_only=False)
    if not isinstance(state, dict):
        raise ValueError(f"{bin_path} does not hold a state dict: {type(state)}")

    rotate, weight, bias = None, None, None
    for key, value in state.items():
        if "rotate_layer" in key:
            if "parametrizations.weight.original" in key or key.endswith(
                "rotate_layer"
            ):
                rotate = value
        elif "learned_source" in key:
            if key.endswith("weight") and "parametrizations" not in key:
                weight = value
            elif key.endswith("bias"):
                bias = value
    if rotate is not None and weight is None:
        # pyreft also saves LoReFT with bare keys (weight/bias alongside
        # rotate_layer); the rotation's presence is what marks LoReFT.
        weight = state.get("weight")
        bias = state.get("bias")
    if rotate is not None:
        if weight is None:
            raise ValueError(
                f"{bin_path} has a rotate_layer but no learned-source "
                f"weight; keys: {sorted(state)}"
            )
        return ReftIntervention(
            rotate_layer=rotate,
            learned_source_weight=weight,
            learned_source_bias=bias,
            layer=layer,
        )

    # BiasIntervention checkpoints store one direction rather than a rotation.
    if len(state) == 1:
        vector = next(iter(state.values()))
    elif "source_representation" in state:
        vector = state["source_representation"]
    elif "bias" in state:
        vector = state["bias"]
    elif "weight" in state:
        vector = state["weight"]
    else:
        raise ValueError(
            f"cannot identify the intervention tensor in {bin_path}; "
            f"keys: {sorted(state)}. Load the checkpoint yourself and "
            "construct a payload directly."
        )
    return DirectionVector({layer: vector})

easysteer.vectors.from_lm_steer

from_lm_steer(
    path: str, vector_index: int = 0
) -> LowRankProjector

Payload from an LM-Steer checkpoint (.pt).

Handles the published gpt2.pt layout (a list whose second entry is the parameter dict). Multi-vector checkpoints stack steer vectors; vector_index selects one and defaults to the first.

Source code in easysteer/vectors.py

def from_lm_steer(path: str, vector_index: int = 0) -> LowRankProjector:
    """Payload from an LM-Steer checkpoint (.pt).

    Handles the published ``gpt2.pt`` layout (a list whose second entry
    is the parameter dict). Multi-vector checkpoints stack steer
    vectors; ``vector_index`` selects one and defaults to the first.
    """
    import torch

    state = torch.load(path, map_location="cpu", weights_only=False)
    if isinstance(state, list) and len(state) > 1:
        state = state[1]
    if not isinstance(state, dict) or not (
        "projector1" in state and "projector2" in state
    ):
        raise ValueError(f"projector matrices not found in {path}")
    p1, p2 = state["projector1"], state["projector2"]
    if p1.dim() > 2:
        if not 0 <= vector_index < p1.shape[0]:
            raise ValueError(
                f"vector_index {vector_index} out of range for a "
                f"{p1.shape[0]}-vector checkpoint"
            )
        p1, p2 = p1[vector_index], p2[vector_index]
    elif vector_index != 0:
        raise ValueError("vector_index given but the checkpoint holds one vector")
    return LowRankProjector(projector1=p1, projector2=p2)

easysteer.vectors.from_linear_transport

from_linear_transport(path: str) -> LinearMap

Payload from a LinearTransport pickle (A_ weight, B_ bias).

Source code in easysteer/vectors.py

def from_linear_transport(path: str) -> LinearMap:
    """Payload from a LinearTransport pickle (``A_`` weight, ``B_`` bias)."""
    with open(path, "rb") as f:
        data = pickle.load(f)
    if isinstance(data, dict):
        weight, bias = data.get("A_"), data.get("B_")
    else:
        weight, bias = getattr(data, "A_", None), getattr(data, "B_", None)
    if weight is None:
        raise ValueError(
            f"weight matrix (A_) not found in {path} (type {type(data).__name__})"
        )
    return LinearMap(weight=weight, bias=bias)