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".	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".
        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
    )

Containers and utilities

easysteer.steer.StatisticalControlVector dataclass

Statistical control vector with multi-layer directions

Source code in easysteer/steer/utils.py

@dataclasses.dataclass
class StatisticalControlVector:
    """Statistical control vector with multi-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 a trained StatisticalControlVector to a llama.cpp .gguf file.
        Compatible with repeng format.
        """
        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):
                    # Handle nested dictionaries like explained_variance
                    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"""
        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 = {}

        # Extract 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 a trained StatisticalControlVector to a llama.cpp .gguf file. Compatible with repeng format.

Source code in easysteer/steer/utils.py

def export_gguf(self, path: os.PathLike[str] | str):
    """
    Export a trained StatisticalControlVector to a llama.cpp .gguf file.
    Compatible with repeng format.
    """
    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):
                # Handle nested dictionaries like explained_variance
                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"""
    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 = {}

    # Extract 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. Never modified or rebound by this function.	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 TRUE 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.
            Never modified or rebound by this function.
        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 TRUE layer ids for CaptureResult
            input and positional indices for nested-list input.
    """
    # CaptureResult (easysteer.hidden_states) is accepted directly; it
    # converts to the nested [sample][layer][token] shape with exact,
    # label-driven per-sample rows, and its TRUE layer ids key the
    # output dicts (legacy nested input keeps positional keys).
    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
                )
                if torch.is_tensor(token_hidden):
                    token_hidden = token_hidden.cpu().float().numpy()
                hiddens[layer_key].append(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: Optional[str] = None) -> List[Dict[str, Any]]

Search for SAE features based on a semantic query

Parameters:

Name	Type	Description	Default
model_id	str	Model identifier (e.g., 'gemma-2-9b')	required
sae_id	str	SAE identifier (e.g., '24-gemmascope-res-16k')	required
query	str	Search query	required
api_key	Optional[str]	Optional API key (will use environment variable if not provided)	None

Returns:

Type	Description
List[Dict[str, Any]]	List of matching features sorted by relevance

Source code in easysteer/steer/sae.py

def search_sae_features(model_id: str, sae_id: str, query: str, api_key: Optional[str] = None) -> List[Dict[str, Any]]:
    """
    Search for SAE features based on a semantic query

    Args:
        model_id: Model identifier (e.g., 'gemma-2-9b')
        sae_id: SAE identifier (e.g., '24-gemmascope-res-16k')
        query: Search query
        api_key: Optional API key (will use environment variable if not provided)

    Returns:
        List of matching features sorted by relevance
    """
    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: Optional[str] = None) -> Dict[str, Any]

Get detailed explanation for a specific feature

Parameters:

Name	Type	Description	Default
model_id	str	Model identifier (e.g., 'gemma-2-9b')	required
sae_id	str	SAE identifier (e.g., '24-gemmascope-res-16k')	required
feature_index	int	Feature index number	required
api_key	Optional[str]	Optional API key (will use environment variable if not provided)	None

Returns:

Type	Description
Dict[str, Any]	Dictionary containing processed feature explanation details

Source code in easysteer/steer/sae.py

def get_sae_feature_explanation(model_id: str, sae_id: str, feature_index: int, api_key: Optional[str] = None) -> Dict[str, Any]:
    """
    Get detailed explanation for a specific feature

    Args:
        model_id: Model identifier (e.g., 'gemma-2-9b')
        sae_id: SAE identifier (e.g., '24-gemmascope-res-16k')
        feature_index: Feature index number
        api_key: Optional API key (will use environment variable if not provided)

    Returns:
        Dictionary containing processed feature explanation details
    """
    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: Optional[str] = None) -> Optional[np.ndarray]

Extract decoder vector for a specific feature index from SAE model file

Parameters:

Name	Type	Description	Default
model_file	str	Path to the SAE model file (npz format)	required
feature_index	int	Feature index to extract	required
save_path	Optional[str]	Optional path to save the vector as PyTorch file (.pt)	None

Returns:

Type	Description
Optional[ndarray]	Decoder vector as numpy array

Source code in easysteer/steer/sae.py

def extract_sae_decoder_vector(model_file: str, feature_index: int, save_path: Optional[str] = None) -> Optional[np.ndarray]:
    """
    Extract decoder vector for a specific feature index from SAE model file

    Args:
        model_file: Path to the SAE model file (npz format)
        feature_index: Feature index to extract
        save_path: Optional path to save the vector as PyTorch file (.pt)

    Returns:
        Decoder vector as numpy 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.from_control_vector

from_control_vector(cv: Any) -> DirectionVector

Payload from an easysteer StatisticalControlVector.

The no-disk path: extract with easysteer.steer and steer with the result directly, no GGUF round-trip.

Source code in easysteer/vectors.py

def from_control_vector(cv: Any) -> DirectionVector:
    """Payload from an easysteer ``StatisticalControlVector``.

    The no-disk path: extract with ``easysteer.steer`` and steer with
    the result directly, no 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 easysteer.steer.utils import StatisticalControlVector

    return from_control_vector(StatisticalControlVector.import_gguf(path))

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: "
            f"{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 :class:DirectionVector for the direct algorithm; a LoReFT state dict (rotation + learned source) becomes a :class: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 :class:`DirectionVector` for the
    ``direct`` algorithm; a LoReFT state dict (rotation + learned
    source) becomes a :class:`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-style: exactly one plausible direction tensor.
    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 — explicitly, instead of the silent first-vector default the engine loader used to apply.

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 — explicitly, instead of the
    silent first-vector default the engine loader used to apply.
    """
    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)