Classification

The classification module provides a backend-agnostic API for training and evaluating text classifiers in Lexos. It exposes a public Classifier façade together with pipeline implementations for spaCy and scikit-learn, plus helpers for saving prediction outputs.

Public API

classification

init.py.

Last Updated: August 30, 2026 Last Tested: August 30, 2026

Modules:

Name	Description
classifier	classifier.py.
sklearn_pipeline	sklearn_pipeline.py.
spacy_pipeline	spacy_pipeline.py.

Classes:

Name	Description
BaseClassificationPipeline	Abstract strategy interface for classification backends.
Classifier	High-level classification orchestration for non-technical users.
ClassifierData	Standardized input wrapper for training and prediction data.
SklearnClassifierPipeline	A scikit-learn-based classification backend.
SpaCyTextCategorizerPipeline	A spaCy TextCategorizer wrapper with a Lexos-friendly API.

BaseClassificationPipeline pydantic-model

Bases: BaseModel

Abstract strategy interface for classification backends.

Subclasses implement the concrete logic for each method, such as spaCy TextCategorizer or a scikit-learn estimator.

Config:

• arbitrary_types_allowed: True

Fields:

• name (str)

Source code in lexos/classification/classifier.py

class BaseClassificationPipeline(BaseModel):
    """Abstract strategy interface for classification backends.

    Subclasses implement the concrete logic for each method, such as spaCy
    `TextCategorizer` or a scikit-learn estimator.
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)
    name: str = Field(default="classifier", description="Human-readable pipeline name.")

    @property
    def model(self) -> Any:
        """Return the underlying backend model object.

        Returns:
            The underlying backend model object.
        """
        raise NotImplementedError

    def fit(self, data: Any, labels: Sequence[str]) -> Any:
        """Train the pipeline on the supplied data and labels.

        Args:
            data: The input data to train on.
            labels: The corresponding labels for the input data.

        Returns:
            The trained pipeline instance.
        """
        raise NotImplementedError

    def predict(self, data: Any) -> Sequence[str]:
        """Predict labels for the supplied data.

        Args:
            data: The input data to make predictions on.

        Returns:
            A list of predicted labels for the input data.
        """
        raise NotImplementedError

    def predict_scores(self, data: Any) -> Sequence[dict[str, float]]:
        """Return the prediction probabilities or confidences for each input item.

        Args:
            data: The input data to make predictions on.

        Returns:
            A list of dictionaries containing prediction probabilities or confidences for each input item.
        """
        raise NotImplementedError

    def evaluate(self, data: Any, labels: Sequence[str]) -> dict[str, float]:
        """Evaluate the fitted pipeline on a dataset.

        Args:
            data: The input data to evaluate on.
            labels: The corresponding labels for the input data.

        Returns:
            A dictionary containing evaluation metrics for the input data and labels.
        """
        raise NotImplementedError

    def save(self, path: str | Any) -> None:
        """Persist the fitted pipeline and its configuration to disk."""
        raise NotImplementedError

    @classmethod
    def load(cls, path: str | Any) -> "BaseClassificationPipeline":
        """Load a saved pipeline instance from disk."""
        raise NotImplementedError

    def __call__(self, data: Any) -> Sequence[str]:
        """Convenience wrapper for predicting on a single data payload."""
        return self.predict(data)

model: Any property

Return the underlying backend model object.

Returns:

Type	Description
Any	The underlying backend model object.

name: str = 'classifier' pydantic-field

Human-readable pipeline name.

__call__(data: Any) -> Sequence[str]

Convenience wrapper for predicting on a single data payload.

Source code in lexos/classification/classifier.py

def __call__(self, data: Any) -> Sequence[str]:
    """Convenience wrapper for predicting on a single data payload."""
    return self.predict(data)

evaluate(data: Any, labels: Sequence[str]) -> dict[str, float]

Evaluate the fitted pipeline on a dataset.

Parameters:

Name	Type	Description	Default
data	Any	The input data to evaluate on.	required
labels	Sequence[str]	The corresponding labels for the input data.	required

Returns:

Type	Description
dict[str, float]	A dictionary containing evaluation metrics for the input data and labels.

Source code in lexos/classification/classifier.py

def evaluate(self, data: Any, labels: Sequence[str]) -> dict[str, float]:
    """Evaluate the fitted pipeline on a dataset.

    Args:
        data: The input data to evaluate on.
        labels: The corresponding labels for the input data.

    Returns:
        A dictionary containing evaluation metrics for the input data and labels.
    """
    raise NotImplementedError

fit(data: Any, labels: Sequence[str]) -> Any

Train the pipeline on the supplied data and labels.

Parameters:

Name	Type	Description	Default
data	Any	The input data to train on.	required
labels	Sequence[str]	The corresponding labels for the input data.	required

Returns:

Type	Description
Any	The trained pipeline instance.

Source code in lexos/classification/classifier.py

def fit(self, data: Any, labels: Sequence[str]) -> Any:
    """Train the pipeline on the supplied data and labels.

    Args:
        data: The input data to train on.
        labels: The corresponding labels for the input data.

    Returns:
        The trained pipeline instance.
    """
    raise NotImplementedError

load(path: str | Any) -> BaseClassificationPipeline classmethod

Load a saved pipeline instance from disk.

Source code in lexos/classification/classifier.py

@classmethod
def load(cls, path: str | Any) -> "BaseClassificationPipeline":
    """Load a saved pipeline instance from disk."""
    raise NotImplementedError

predict(data: Any) -> Sequence[str]

Predict labels for the supplied data.

Parameters:

Name	Type	Description	Default
data	Any	The input data to make predictions on.	required

Returns:

Type	Description
Sequence[str]	A list of predicted labels for the input data.

Source code in lexos/classification/classifier.py

def predict(self, data: Any) -> Sequence[str]:
    """Predict labels for the supplied data.

    Args:
        data: The input data to make predictions on.

    Returns:
        A list of predicted labels for the input data.
    """
    raise NotImplementedError

predict_scores(data: Any) -> Sequence[dict[str, float]]

Return the prediction probabilities or confidences for each input item.

Parameters:

Name	Type	Description	Default
data	Any	The input data to make predictions on.	required

Returns:

Type	Description
Sequence[dict[str, float]]	A list of dictionaries containing prediction probabilities or confidences for each input item.

Source code in lexos/classification/classifier.py

def predict_scores(self, data: Any) -> Sequence[dict[str, float]]:
    """Return the prediction probabilities or confidences for each input item.

    Args:
        data: The input data to make predictions on.

    Returns:
        A list of dictionaries containing prediction probabilities or confidences for each input item.
    """
    raise NotImplementedError

save(path: str | Any) -> None

Persist the fitted pipeline and its configuration to disk.

Source code in lexos/classification/classifier.py

def save(self, path: str | Any) -> None:
    """Persist the fitted pipeline and its configuration to disk."""
    raise NotImplementedError

Classifier pydantic-model

Bases: BaseModel

High-level classification orchestration for non-technical users.

The Classifier class is intentionally backend-agnostic. Users supply data, labels, and a pipeline object that implements the backend-specific training and inference logic. This keeps the public API stable across SpaCy, scikit-learn, and other classification methods.

Config:

• arbitrary_types_allowed: True

Fields:

• data (Any | None)
• labels (Sequence[Any])
• titles (Sequence[Any])
• pipeline (BaseClassificationPipeline | None)

Validators:

• _validate_pipeline

Source code in lexos/classification/classifier.py

class Classifier(BaseModel):
    """High-level classification orchestration for non-technical users.

    The `Classifier` class is intentionally backend-agnostic. Users supply data,
    labels, and a pipeline object that implements the backend-specific training and
    inference logic. This keeps the public API stable across SpaCy, scikit-learn,
    and other classification methods.
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)

    data: Any | None = Field(default=None, description="Training or prediction data.")
    labels: Sequence[Any] = Field(
        default_factory=list, description="Classification labels."
    )
    titles: Sequence[Any] = Field(
        default_factory=list, description="Document titles aligned with the data rows."
    )
    pipeline: BaseClassificationPipeline | None = Field(
        default=None,
        description="Classification backend strategy; e.g. a spaCy or scikit-learn pipeline.",
    )

    _fitted: bool = PrivateAttr(default=False)
    _last_fit_data: Any = PrivateAttr(default=None)
    _last_fit_labels: list[Any] = PrivateAttr(default_factory=list)

    @model_validator(mode="after")
    def _validate_pipeline(self) -> "Classifier":
        """Validate that the pipeline is correctly set and an instance of BaseClassificationPipeline.

        Returns:
            The validated Classifier instance.

        Raises:
            TypeError: If the pipeline is not an instance of BaseClassificationPipeline.
        """
        if self.pipeline is not None and not isinstance(
            self.pipeline, BaseClassificationPipeline
        ):
            raise TypeError(
                "pipeline must be an instance of BaseClassificationPipeline."
            )
        return self

    def _resolve_data_and_labels(
        self,
        data: Any | None = None,
        labels: Sequence[Any] | None = None,
        titles: Sequence[Any] | None = None,
    ) -> tuple[ClassifierData, list[Any]]:
        """Resolve the input data and labels, normalizing them into a ClassifierData instance.

        Args:
            data: The input data to resolve. If None, the stored data is used.
            labels: The input labels to resolve. If None, the stored labels are used.
            titles: Optional titles aligned with the input rows.

        Returns:
            A tuple containing the normalized ClassifierData instance and the corresponding list of labels.
        """
        resolved_data = self.data if data is None else data
        explicit_labels = list(labels) if labels is not None else None
        if explicit_labels is None:
            explicit_labels = list(self.labels) if self.labels else None

        explicit_titles = list(titles) if titles is not None else None
        normalized = ClassifierData.from_input(
            resolved_data,
            explicit_labels,
            titles=explicit_titles,
        )

        if explicit_titles is None and self.titles:
            if len(self.titles) == normalized.row_count():
                normalized.titles = list(self.titles)

        return normalized, normalized.labels

    def fit(
        self, data: Any | None = None, labels: Sequence[Any] | None = None
    ) -> "Classifier":
        """Fit the classifier using the supplied pipeline backend.

        Args:
            data: The input data to fit the classifier on. If None, the stored data is used.
            labels: The corresponding labels for the input data. If None, the stored labels are used.

        Returns:
            The fitted Classifier instance.
        """
        if self.pipeline is None:
            raise ValueError(
                "A classification pipeline must be configured before calling fit()."
            )

        resolved_data, resolved_labels = self._resolve_data_and_labels(data, labels)
        self.pipeline.fit(resolved_data, resolved_labels)
        self.data = resolved_data.values
        self.labels = resolved_labels
        self.titles = (
            list(resolved_data.titles) if resolved_data.titles is not None else []
        )
        self._fitted = True
        self._last_fit_data = resolved_data
        self._last_fit_labels = resolved_labels
        return self

    def predict(self, data: Any | None = None) -> list[str]:
        """Predict labels for the supplied data or for the stored training data.

        Args:
            data: The input data to predict labels for. If None, the stored data is used.

        Returns:
            A list of predicted labels for each document.
        """
        if self.pipeline is None:
            raise ValueError(
                "A classification pipeline must be configured before calling predict()."
            )

        if data is None:
            if self.data is None:
                raise ValueError(
                    "No prediction data was supplied and no fitted data is available."
                )
            data = self.data

        predictions = self.pipeline.predict(data)
        return [
            value if isinstance(value, (list, tuple, set)) else str(value)
            for value in predictions
        ]

    def predict_scores(self, data: Any | None = None) -> list[dict[str, float]]:
        """Return the underlying confidence scores for each prediction when available.

        Args:
            data: The input data to predict scores for. If None, the stored data is used.

        Returns:
            A list of dictionaries containing confidence scores for each prediction.
        """
        if self.pipeline is None:
            raise ValueError(
                "A classification pipeline must be configured before calling predict_scores()."
            )

        if data is None:
            if self.data is None:
                raise ValueError(
                    "No prediction data was supplied and no fitted data is available."
                )
            data = self.data

        if not hasattr(self.pipeline, "predict_scores"):
            raise NotImplementedError(
                f"{type(self.pipeline).__name__} does not implement predict_scores()."
            )

        scores = self.pipeline.predict_scores(data)
        # Sort the scores
        for row in scores:
            row.update(
                dict(sorted(row.items(), key=lambda item: item[1], reverse=True))
            )
        return scores

    def evaluate(
        self, data: Any | None = None, labels: Sequence[Any] | None = None
    ) -> dict[str, float]:
        """Evaluate the fitted pipeline on the supplied data and labels.

        Args:
            data: The input data to evaluate the pipeline on. If None, the stored data is used.
            labels: The corresponding labels for the input data. If None, the stored labels are used.

        Returns:
            A dictionary containing evaluation metrics for the predictions.
        """
        if self.pipeline is None:
            raise ValueError(
                "A classification pipeline must be configured before calling evaluate()."
            )

        if data is None:
            if self.data is None:
                raise ValueError(
                    "No evaluation data was supplied and no fitted data is available."
                )
            data = self.data
        if labels is None:
            if not self.labels:
                raise ValueError("No labels were provided for evaluation.")
            labels = self.labels

        return self.pipeline.evaluate(data, list(labels))

    def split_data(
        self,
        data: Any | None = None,
        labels: Sequence[str] | None = None,
        titles: Sequence[Any] | None = None,
        test_size: float = 0.2,
        dev_size: float | None = None,
        random_state: int = 42,
        stratify: bool = True,
    ) -> dict[str, Any]:
        """Split data into train/test/dev partitions.

        Args:
            data: data to split; defaults to the classifier's stored data.
            labels: labels aligned to the data; defaults to the classifier's labels.
            titles: optional titles aligned to the rows; preserved in the output.
            test_size: fraction of the data reserved for testing.
            dev_size: optional fraction reserved for development / validation.
            random_state: deterministic random seed.
            stratify: whether to preserve label distributions across splits.

        Returns:
            Dictionary containing the partitions keyed by `train`, `test`, and optional
            `dev` data plus the corresponding labels and titles.
        """
        resolved_data, resolved_labels = self._resolve_data_and_labels(
            data,
            labels,
            titles=titles,
        )

        split = resolved_data.split(
            test_size=test_size,
            dev_size=dev_size,
            random_state=random_state,
            stratify=stratify,
        )

        result = {
            "train": {
                "data": split["train"].values,
                "labels": split["train"].labels,
                "titles": split["train"].titles,
            },
            "test": {
                "data": split["test"].values,
                "labels": split["test"].labels,
                "titles": split["test"].titles,
            },
        }
        if "dev" in split:
            result["dev"] = {
                "data": split["dev"].values,
                "labels": split["dev"].labels,
                "titles": split["dev"].titles,
            }
        return result

    def train_test_split(
        self,
        test_size: float = 0.2,
        random_state: int = 42,
        stratify: bool = True,
        titles: Sequence[Any] | None = None,
    ) -> dict[str, Any]:
        """Convenience wrapper for train/test splitting.

        Args:
            test_size: fraction of the data reserved for testing.
            random_state: deterministic random seed.
            stratify: whether to preserve label distributions across splits.
            titles: optional titles aligned with the rows; preserved in the output.

        Returns:
            Dictionary containing the train and test partitions keyed by `train` and `test` data plus the corresponding labels and titles.
        """
        return self.split_data(
            titles=titles,
            test_size=test_size,
            random_state=random_state,
            stratify=stratify,
        )

    def train_dev_split(
        self,
        dev_size: float = 0.2,
        random_state: int = 42,
        stratify: bool = True,
        titles: Sequence[Any] | None = None,
    ) -> dict[str, Any]:
        """Convenience wrapper for train/dev splitting.

        Args:
            dev_size: fraction of the data reserved for development / validation.
            random_state: deterministic random seed.
            stratify: whether to preserve label distributions across splits.
            titles: optional titles aligned with the rows; preserved in the output.

        Returns:
            Dictionary containing the train and dev partitions keyed by `train` and `dev` data plus the corresponding labels and titles.
        """
        return self.split_data(
            titles=titles,
            dev_size=dev_size,
            random_state=random_state,
            stratify=stratify,
        )

    def __call__(self, data: Any) -> list[str]:
        """Convenience wrapper for prediction calls.

        Args:
            data: The input data to make predictions on.

        Returns:
            A list of predicted labels for the input data.
        """
        return self.predict(data)

data: Any | None = None pydantic-field

Training or prediction data.

labels: Sequence[Any] pydantic-field

Classification labels.

pipeline: BaseClassificationPipeline | None = None pydantic-field

Classification backend strategy; e.g. a spaCy or scikit-learn pipeline.

titles: Sequence[Any] pydantic-field

Document titles aligned with the data rows.

__call__(data: Any) -> list[str]

Convenience wrapper for prediction calls.

Parameters:

Name	Type	Description	Default
data	Any	The input data to make predictions on.	required

Returns:

Type	Description
list[str]	A list of predicted labels for the input data.

Source code in lexos/classification/classifier.py

def __call__(self, data: Any) -> list[str]:
    """Convenience wrapper for prediction calls.

    Args:
        data: The input data to make predictions on.

    Returns:
        A list of predicted labels for the input data.
    """
    return self.predict(data)

evaluate(data: Any | None = None, labels: Sequence[Any] | None = None) -> dict[str, float]

Evaluate the fitted pipeline on the supplied data and labels.

Parameters:

Name	Type	Description	Default
data	Any | None	The input data to evaluate the pipeline on. If None, the stored data is used.	None
labels	Sequence[Any] | None	The corresponding labels for the input data. If None, the stored labels are used.	None

Returns:

Type	Description
dict[str, float]	A dictionary containing evaluation metrics for the predictions.

Source code in lexos/classification/classifier.py

def evaluate(
    self, data: Any | None = None, labels: Sequence[Any] | None = None
) -> dict[str, float]:
    """Evaluate the fitted pipeline on the supplied data and labels.

    Args:
        data: The input data to evaluate the pipeline on. If None, the stored data is used.
        labels: The corresponding labels for the input data. If None, the stored labels are used.

    Returns:
        A dictionary containing evaluation metrics for the predictions.
    """
    if self.pipeline is None:
        raise ValueError(
            "A classification pipeline must be configured before calling evaluate()."
        )

    if data is None:
        if self.data is None:
            raise ValueError(
                "No evaluation data was supplied and no fitted data is available."
            )
        data = self.data
    if labels is None:
        if not self.labels:
            raise ValueError("No labels were provided for evaluation.")
        labels = self.labels

    return self.pipeline.evaluate(data, list(labels))

fit(data: Any | None = None, labels: Sequence[Any] | None = None) -> Classifier

Fit the classifier using the supplied pipeline backend.

Parameters:

Name	Type	Description	Default
data	Any | None	The input data to fit the classifier on. If None, the stored data is used.	None
labels	Sequence[Any] | None	The corresponding labels for the input data. If None, the stored labels are used.	None

Returns:

Type	Description
Classifier	The fitted Classifier instance.

Source code in lexos/classification/classifier.py

def fit(
    self, data: Any | None = None, labels: Sequence[Any] | None = None
) -> "Classifier":
    """Fit the classifier using the supplied pipeline backend.

    Args:
        data: The input data to fit the classifier on. If None, the stored data is used.
        labels: The corresponding labels for the input data. If None, the stored labels are used.

    Returns:
        The fitted Classifier instance.
    """
    if self.pipeline is None:
        raise ValueError(
            "A classification pipeline must be configured before calling fit()."
        )

    resolved_data, resolved_labels = self._resolve_data_and_labels(data, labels)
    self.pipeline.fit(resolved_data, resolved_labels)
    self.data = resolved_data.values
    self.labels = resolved_labels
    self.titles = (
        list(resolved_data.titles) if resolved_data.titles is not None else []
    )
    self._fitted = True
    self._last_fit_data = resolved_data
    self._last_fit_labels = resolved_labels
    return self

predict(data: Any | None = None) -> list[str]

Predict labels for the supplied data or for the stored training data.

Parameters:

Name	Type	Description	Default
data	Any | None	The input data to predict labels for. If None, the stored data is used.	None

Returns:

Type	Description
list[str]	A list of predicted labels for each document.

Source code in lexos/classification/classifier.py

def predict(self, data: Any | None = None) -> list[str]:
    """Predict labels for the supplied data or for the stored training data.

    Args:
        data: The input data to predict labels for. If None, the stored data is used.

    Returns:
        A list of predicted labels for each document.
    """
    if self.pipeline is None:
        raise ValueError(
            "A classification pipeline must be configured before calling predict()."
        )

    if data is None:
        if self.data is None:
            raise ValueError(
                "No prediction data was supplied and no fitted data is available."
            )
        data = self.data

    predictions = self.pipeline.predict(data)
    return [
        value if isinstance(value, (list, tuple, set)) else str(value)
        for value in predictions
    ]

predict_scores(data: Any | None = None) -> list[dict[str, float]]

Return the underlying confidence scores for each prediction when available.

Parameters:

Name	Type	Description	Default
data	Any | None	The input data to predict scores for. If None, the stored data is used.	None

Returns:

Type	Description
list[dict[str, float]]	A list of dictionaries containing confidence scores for each prediction.

Source code in lexos/classification/classifier.py

def predict_scores(self, data: Any | None = None) -> list[dict[str, float]]:
    """Return the underlying confidence scores for each prediction when available.

    Args:
        data: The input data to predict scores for. If None, the stored data is used.

    Returns:
        A list of dictionaries containing confidence scores for each prediction.
    """
    if self.pipeline is None:
        raise ValueError(
            "A classification pipeline must be configured before calling predict_scores()."
        )

    if data is None:
        if self.data is None:
            raise ValueError(
                "No prediction data was supplied and no fitted data is available."
            )
        data = self.data

    if not hasattr(self.pipeline, "predict_scores"):
        raise NotImplementedError(
            f"{type(self.pipeline).__name__} does not implement predict_scores()."
        )

    scores = self.pipeline.predict_scores(data)
    # Sort the scores
    for row in scores:
        row.update(
            dict(sorted(row.items(), key=lambda item: item[1], reverse=True))
        )
    return scores

split_data(data: Any | None = None, labels: Sequence[str] | None = None, titles: Sequence[Any] | None = None, test_size: float = 0.2, dev_size: float | None = None, random_state: int = 42, stratify: bool = True) -> dict[str, Any]

Split data into train/test/dev partitions.

Parameters:

Name	Type	Description	Default
data	Any | None	data to split; defaults to the classifier's stored data.	None
labels	Sequence[str] | None	labels aligned to the data; defaults to the classifier's labels.	None
titles	Sequence[Any] | None	optional titles aligned to the rows; preserved in the output.	None
test_size	float	fraction of the data reserved for testing.	0.2
dev_size	float | None	optional fraction reserved for development / validation.	None
random_state	int	deterministic random seed.	42
stratify	bool	whether to preserve label distributions across splits.	True

Returns:

Type	Description
dict[str, Any]	Dictionary containing the partitions keyed by train, test, and optional
dict[str, Any]	dev data plus the corresponding labels and titles.

Source code in lexos/classification/classifier.py

def split_data(
    self,
    data: Any | None = None,
    labels: Sequence[str] | None = None,
    titles: Sequence[Any] | None = None,
    test_size: float = 0.2,
    dev_size: float | None = None,
    random_state: int = 42,
    stratify: bool = True,
) -> dict[str, Any]:
    """Split data into train/test/dev partitions.

    Args:
        data: data to split; defaults to the classifier's stored data.
        labels: labels aligned to the data; defaults to the classifier's labels.
        titles: optional titles aligned to the rows; preserved in the output.
        test_size: fraction of the data reserved for testing.
        dev_size: optional fraction reserved for development / validation.
        random_state: deterministic random seed.
        stratify: whether to preserve label distributions across splits.

    Returns:
        Dictionary containing the partitions keyed by `train`, `test`, and optional
        `dev` data plus the corresponding labels and titles.
    """
    resolved_data, resolved_labels = self._resolve_data_and_labels(
        data,
        labels,
        titles=titles,
    )

    split = resolved_data.split(
        test_size=test_size,
        dev_size=dev_size,
        random_state=random_state,
        stratify=stratify,
    )

    result = {
        "train": {
            "data": split["train"].values,
            "labels": split["train"].labels,
            "titles": split["train"].titles,
        },
        "test": {
            "data": split["test"].values,
            "labels": split["test"].labels,
            "titles": split["test"].titles,
        },
    }
    if "dev" in split:
        result["dev"] = {
            "data": split["dev"].values,
            "labels": split["dev"].labels,
            "titles": split["dev"].titles,
        }
    return result

train_dev_split(dev_size: float = 0.2, random_state: int = 42, stratify: bool = True, titles: Sequence[Any] | None = None) -> dict[str, Any]

Convenience wrapper for train/dev splitting.

Parameters:

Name	Type	Description	Default
dev_size	float	fraction of the data reserved for development / validation.	0.2
random_state	int	deterministic random seed.	42
stratify	bool	whether to preserve label distributions across splits.	True
titles	Sequence[Any] | None	optional titles aligned with the rows; preserved in the output.	None

Returns:

Type	Description
dict[str, Any]	Dictionary containing the train and dev partitions keyed by train and dev data plus the corresponding labels and titles.

Source code in lexos/classification/classifier.py

def train_dev_split(
    self,
    dev_size: float = 0.2,
    random_state: int = 42,
    stratify: bool = True,
    titles: Sequence[Any] | None = None,
) -> dict[str, Any]:
    """Convenience wrapper for train/dev splitting.

    Args:
        dev_size: fraction of the data reserved for development / validation.
        random_state: deterministic random seed.
        stratify: whether to preserve label distributions across splits.
        titles: optional titles aligned with the rows; preserved in the output.

    Returns:
        Dictionary containing the train and dev partitions keyed by `train` and `dev` data plus the corresponding labels and titles.
    """
    return self.split_data(
        titles=titles,
        dev_size=dev_size,
        random_state=random_state,
        stratify=stratify,
    )

train_test_split(test_size: float = 0.2, random_state: int = 42, stratify: bool = True, titles: Sequence[Any] | None = None) -> dict[str, Any]

Convenience wrapper for train/test splitting.

Parameters:

Name	Type	Description	Default
test_size	float	fraction of the data reserved for testing.	0.2
random_state	int	deterministic random seed.	42
stratify	bool	whether to preserve label distributions across splits.	True
titles	Sequence[Any] | None	optional titles aligned with the rows; preserved in the output.	None

Returns:

Type	Description
dict[str, Any]	Dictionary containing the train and test partitions keyed by train and test data plus the corresponding labels and titles.

Source code in lexos/classification/classifier.py

def train_test_split(
    self,
    test_size: float = 0.2,
    random_state: int = 42,
    stratify: bool = True,
    titles: Sequence[Any] | None = None,
) -> dict[str, Any]:
    """Convenience wrapper for train/test splitting.

    Args:
        test_size: fraction of the data reserved for testing.
        random_state: deterministic random seed.
        stratify: whether to preserve label distributions across splits.
        titles: optional titles aligned with the rows; preserved in the output.

    Returns:
        Dictionary containing the train and test partitions keyed by `train` and `test` data plus the corresponding labels and titles.
    """
    return self.split_data(
        titles=titles,
        test_size=test_size,
        random_state=random_state,
        stratify=stratify,
    )

ClassifierData pydantic-model

Bases: BaseModel

Standardized input wrapper for training and prediction data.

This object centralizes the data-shape concerns ensuring that data is handled consistently before it is passed to Classifier.

Config:

• arbitrary_types_allowed: True

Fields:

• values (Any)
• labels (list[Any])
• docs (Any)
• titles (list[Any] | None)
• matrix (Any)
• source (str)

Source code in lexos/classification/classifier.py

class ClassifierData(BaseModel):
    """Standardized input wrapper for training and prediction data.

    This object centralizes the data-shape concerns ensuring that data is handled consistently before it is passed to `Classifier`.
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)

    values: Any
    labels: list[Any] = Field(default_factory=list)
    docs: Any = None
    titles: list[Any] | None = None
    matrix: Any = None
    source: str = "raw"

    def __init__(
        self,
        values: Any,
        labels: Sequence[Any] | None = None,
        *,
        docs: Any = None,
        titles: Sequence[Any] | None = None,
        matrix: Any = None,
        source: str = "raw",
    ) -> None:
        """Initialize the ClassifierData object.

        Args:
            values: The main data values.
            labels: Optional sequence of labels corresponding to the data.
            docs: Optional sequence of document objects.
            titles: Optional sequence of titles for the data items.
            matrix: Optional matrix representation of the data.
            source: A string indicating the source of the data.
        """
        super().__init__(
            values=values,
            labels=list(labels) if labels is not None else [],
            docs=docs,
            titles=list(titles) if titles is not None else None,
            matrix=matrix,
            source=source,
        )

    @staticmethod
    def _matrix_row_count(matrix: Any) -> int:
        """Return the number of rows in a matrix-like object, including list-backed inputs.

        Args:
            matrix: The matrix-like object to count rows for.

        Returns:
            The number of rows in the matrix-like object.
        """
        shape = getattr(matrix, "shape", None)
        if shape is not None:
            return int(shape[0])
        return len(matrix)

    @staticmethod
    def _validate_label_count(
        expected_count: int | None, actual_count: int, context: str
    ) -> None:
        """Raise a ValueError when label length diverges from the data shape.

        Args:
            expected_count: The expected number of labels.
            actual_count: The actual number of labels.
            context: A message to include in the ValueError if the counts do not match.

        Raises:
            ValueError: If the expected count is not None and does not match the actual count.
        """
        if expected_count is not None and expected_count != actual_count:
            raise ValueError(f"{context}")

    @classmethod
    def _from_dtm_input(
        cls,
        data: Any,
        labels: Sequence[Any] | None = None,
        titles: Sequence[Any] | None = None,
    ) -> "ClassifierData":
        """Normalize a Lexos DTM object into a standardized Dataset wrapper.

        Args:
            data: The Lexos DTM object to normalize.
            labels: Optional sequence of labels to associate with the data. If not provided, labels will be inferred from the DTM object.
            titles: Optional sequence of titles aligned with the records.

        Returns:
            An instance of the ClassifierData class wrapping the normalized data.
        """
        matrix = getattr(data, "doc_term_matrix", None)
        docs = getattr(data, "docs", None)
        resolved_labels = list(getattr(data, "labels", []) or [])
        resolved_titles = (
            list(getattr(data, "titles", []) or []) if hasattr(data, "titles") else []
        )
        if labels is not None and len(labels) > 0:
            resolved_labels = list(labels)
        if titles is not None and len(titles) > 0:
            resolved_titles = list(titles)

        if matrix is not None:
            row_count = cls._matrix_row_count(matrix)
        elif docs is not None:
            row_count = len(docs)
        else:
            row_count = len(resolved_labels)

        resolved_labels, resolved_titles = cls._resolve_row_alignment(
            row_count,
            resolved_labels,
            resolved_titles,
            "Label count must match the row count in the DTM document-term matrix."
            if matrix is not None
            else "Label count must match the number of stored DTM docs.",
            "Title count must match the number of rows in the dataset.",
        )

        return cls(
            values=matrix if matrix is not None else docs,
            labels=resolved_labels,
            docs=docs,
            titles=resolved_titles,
            matrix=matrix,
            source="dtm",
        )

    @classmethod
    def _from_dataframe_input(
        cls,
        data: pd.DataFrame,
        labels: Sequence[Any] | None = None,
        titles: Sequence[Any] | None = None,
    ) -> "ClassifierData":
        """Normalize a pandas DataFrame input into a standardized Dataset wrapper.

        Args:
            data: The pandas DataFrame to normalize.
            labels: Optional sequence of labels to associate with the data. If not provided, labels will be inferred from the DataFrame.

        Returns:
            An instance of the ClassifierData class wrapping the normalized data.
        """
        frame_labels = (
            list(data["label"].astype(str)) if "label" in data.columns else []
        )
        resolved_labels = list(frame_labels if labels is None else labels)
        resolved_titles = list(data["title"]) if "title" in data.columns else []
        if titles is not None and len(titles) > 0:
            resolved_titles = list(titles)

        resolved_labels, resolved_titles = cls._resolve_row_alignment(
            len(data),
            resolved_labels,
            resolved_titles,
            "Label count must match the row count in the DataFrame.",
            "Title count must match the row count in the DataFrame.",
        )

        return cls(
            values=data,
            labels=resolved_labels,
            titles=resolved_titles,
            source="dataframe",
        )

    @classmethod
    def _from_matrix_input(
        cls,
        data: Any,
        labels: Sequence[Any] | None = None,
        titles: Sequence[Any] | None = None,
    ) -> "ClassifierData":
        """Normalize matrix-like inputs into a standardized Dataset wrapper.

        Args:
            data: The matrix-like data to normalize.
            labels: Optional sequence of labels to associate with the data. If not provided, labels will be inferred from the matrix-like data.

        Returns:
            An instance of the ClassifierData class wrapping the normalized data.
        """
        resolved_labels = list(labels or [])
        resolved_titles = list(titles) if titles is not None else []
        row_count = cls._matrix_row_count(data)

        resolved_labels, resolved_titles = cls._resolve_row_alignment(
            row_count,
            resolved_labels,
            resolved_titles,
            "Label count must match the row count in the matrix-like data.",
            "Title count must match the row count in the matrix-like data.",
        )

        return cls(
            values=data,
            labels=resolved_labels,
            titles=resolved_titles,
            matrix=data,
            source="matrix",
        )

    @classmethod
    def _from_sequence_input(
        cls,
        data: Sequence[Any],
        labels: Sequence[Any] | None = None,
        titles: Sequence[Any] | None = None,
    ) -> "ClassifierData":
        """Normalize native Python sequences into a standardized Dataset wrapper.

        Args:
            data: The sequence of data items to normalize.
            labels: Optional sequence of labels to associate with the data. If not provided, labels will be inferred from the sequence.

        Returns:
            An instance of the ClassifierData class wrapping the normalized data.
        """
        resolved_labels = list(labels or [])
        resolved_titles = list(titles) if titles is not None else []

        resolved_labels, resolved_titles = cls._resolve_row_alignment(
            len(data),
            resolved_labels,
            resolved_titles,
            "Number of labels must match the number of data items.",
            "Title count must match the number of data items.",
        )

        return cls(
            values=list(data),
            labels=resolved_labels,
            titles=resolved_titles,
            source="raw",
        )

    @staticmethod
    def _resolve_row_alignment(
        row_count: int,
        labels: Sequence[Any] | None,
        titles: Sequence[Any] | None,
        labels_message: str,
        titles_message: str,
    ) -> tuple[list[Any], list[Any] | None]:
        """Validate label and title counts against a row count and return normalized values."""
        resolved_labels = list(labels or [])
        resolved_titles = list(titles) if titles is not None else []

        ClassifierData._validate_label_count(
            len(resolved_labels),
            row_count,
            labels_message,
        )
        if resolved_titles:
            ClassifierData._validate_label_count(
                len(resolved_titles),
                row_count,
                titles_message,
            )

        return resolved_labels, resolved_titles or None

    @classmethod
    def from_input(
        cls,
        data: Any,
        labels: Sequence[Any] | None = None,
        titles: Sequence[Any] | None = None,
    ) -> "ClassifierData":
        """Normalize raw text, DataFrames, and Lexos DTM objects to a standard form.

        Args:
            data: The input data to normalize. Can be raw text, a pandas DataFrame, a Lexos DTM object, or a matrix-like structure.
            labels: Optional sequence of labels to associate with the data. If not provided, labels will be inferred from the input data.
            titles: Optional sequence of titles aligned with the rows.

        Returns:
            An instance of the ClassifierData class wrapping the normalized data.
        """
        if _is_dtm_like(data):
            return cls._from_dtm_input(data, labels, titles)
        if isinstance(data, pd.DataFrame):
            return cls._from_dataframe_input(data, labels, titles)
        if hasattr(data, "shape") and getattr(data, "ndim", None) == 2:
            return cls._from_matrix_input(data, labels, titles)
        if isinstance(data, (list, tuple)):
            if data and _is_spacy_doc(data[0]):
                resolved_labels, resolved_titles = cls._resolve_row_alignment(
                    len(data),
                    labels,
                    titles,
                    "Number of labels must match the number of data items.",
                    "Title count must match the number of data items.",
                )
                return cls(
                    values=list(data),
                    labels=resolved_labels,
                    titles=resolved_titles,
                    docs=list(data),
                    source="raw",
                )
            return cls._from_sequence_input(data, labels, titles)
        if data is None:
            raise ValueError("No data was supplied to the classifier.")

        return cls(
            values=data,
            labels=list(labels or []),
            titles=list(titles) if titles is not None else None,
            source="raw",
        )

    def row_count(self) -> int:
        """Return the number of rows represented by the standardized input.

        Returns:
            The number of rows represented by the standardized input.
        """
        if self.matrix is not None:
            return self.matrix.shape[0]
        if self.docs is not None:
            return len(self.docs)
        if isinstance(self.values, pd.DataFrame):
            return len(self.values)
        if isinstance(self.values, (list, tuple)):
            return len(self.values)
        return len(self.labels)

    @staticmethod
    def _as_text_for_doc(doc: Any) -> str:
        """Convert a document-like object to text while preserving original tokenization.

        Args:
            doc: The document-like object to convert to text. Can be a spaCy Doc, an object with a `text` attribute, or a sequence of items.

        Returns:
            The text representation of the document-like object.
        """
        if doc is None:
            return ""

        module_name = type(doc).__module__ if doc is not None else ""
        if (
            module_name.startswith("spacy.")
            and hasattr(doc, "vocab")
            and hasattr(doc, "__iter__")
        ):
            return " ".join(token.text for token in doc)
        if hasattr(doc, "text") and not (
            hasattr(doc, "vocab") and hasattr(doc, "__iter__")
        ):
            return str(doc.text)
        if isinstance(doc, (list, tuple, set)):
            return " ".join(str(item) for item in doc)
        return str(doc)

    def as_texts(self) -> list[str]:
        """Return the data as plain text strings when possible.

        Returns:
            A list of plain text strings representing the data.
        """
        if self.docs is not None:
            texts: list[str] = []
            for doc in self.docs:
                texts.append(self._as_text_for_doc(doc))
            return texts

        if isinstance(self.values, pd.DataFrame):
            return [str(item) for item in self.values.to_dict(orient="records")]

        if isinstance(self.values, (list, tuple)):
            texts: list[str] = []
            for item in self.values:
                if _is_spacy_doc(item):
                    texts.append(" ".join(token.text for token in item))
                else:
                    texts.append(str(item))
            return texts

        return [str(self.values)]

    def subset(self, indices: Sequence[int]) -> "ClassifierData":
        """Return a new data object containing only the selected row indices.

        Args:
            indices: A sequence of row indices to include in the subset.

        Returns:
            A new `ClassifierData` object containing only the selected rows.
        """
        idx_list = list(indices)
        selected_titles = (
            [self.titles[i] for i in idx_list] if self.titles is not None else None
        )
        if self.matrix is not None:
            return ClassifierData(
                values=self.matrix[idx_list],
                labels=[self.labels[i] for i in idx_list],
                docs=[self.docs[i] for i in idx_list]
                if self.docs is not None
                else None,
                titles=selected_titles,
                matrix=self.matrix[idx_list],
                source=self.source,
            )

        if isinstance(self.values, pd.DataFrame):
            data = self.values.iloc[idx_list]
            return ClassifierData(
                values=data,
                labels=[self.labels[i] for i in idx_list],
                titles=selected_titles,
                source=self.source,
            )

        sliced = [self.values[i] for i in idx_list]
        return ClassifierData(
            values=sliced,
            labels=[self.labels[i] for i in idx_list],
            titles=selected_titles,
            source=self.source,
        )

    @staticmethod
    def _test_count_for_group(group_size: int, test_size: float) -> int:
        """Compute the number of rows to reserve for testing within one label group.

        Args:
            group_size: The number of rows in the label group.
            test_size: The proportion of the group to reserve for testing.

        Returns:
            The number of rows to reserve for testing within the group.
        """
        if group_size <= 1:
            return 0

        count = int(round(group_size * test_size))
        if count >= group_size:
            count = group_size - 1
        if count <= 0:
            count = 1
        return count

    @staticmethod
    def _dev_count_for_group(group_size: int, dev_size: float) -> int:
        """Compute the number of rows to reserve for development within a train split.

        Args:
            group_size: The number of rows in the train split.
            dev_size: The proportion of the train split to reserve for development.

        Returns:
            The number of rows to reserve for development within the train split.
        """
        if group_size <= 1:
            return 0

        count = int(round(group_size * dev_size))
        if count >= group_size:
            count = max(0, group_size - 1)
        if count <= 0:
            count = 1 if group_size > 1 else 0
        return count

    def _group_label_indices(self) -> dict[str, list[int]]:
        """Group row indices by label value for stratified splitting.

        Returns:
            A dictionary mapping each label value to a list of row indices that have that label.
        """
        grouped: dict[str, list[int]] = defaultdict(list)
        for idx, label in enumerate(self.labels):
            grouped[str(label)].append(idx)
        return grouped

    def _split_indices_by_labels(
        self,
        test_size: float,
        dev_size: float | None,
        random_state: int,
    ) -> dict[str, list[int]]:
        """Split row indices while preserving each class distribution.

        Args:
            test_size: The proportion of the dataset to reserve for testing.
            dev_size: The proportion of the training set to reserve for development, or None if no development set is needed.
            random_state: The seed for the random number generator to ensure reproducibility.

        Returns:
            A dictionary containing the split row indices with keys "train", "test", and optionally "dev".
        """
        grouped = self._group_label_indices()
        train_indices: list[int] = []
        test_indices: list[int] = []
        rng = random.Random(random_state)

        for label_indices in grouped.values():
            rng.shuffle(label_indices)
            label_test_count = self._test_count_for_group(len(label_indices), test_size)
            test_indices.extend(label_indices[:label_test_count])
            train_indices.extend(label_indices[label_test_count:])

        rng.shuffle(train_indices)
        rng.shuffle(test_indices)

        result = {"train": train_indices, "test": test_indices}
        if dev_size is not None:
            if len(train_indices) <= 1:
                result["dev"] = []
            else:
                dev_count = self._dev_count_for_group(len(train_indices), dev_size)
                dev_indices = train_indices[:dev_count]
                result["dev"] = dev_indices
                result["train"] = [
                    idx for idx in train_indices if idx not in set(dev_indices)
                ]
        return result

    @staticmethod
    def _validate_split_parameters(
        n_rows: int, test_size: float, dev_size: float | None
    ) -> None:
        """Validate dataset split parameters before partitioning.

        Args:
            n_rows: The total number of rows in the dataset.
            test_size: The proportion of the dataset to reserve for testing.
            dev_size: The proportion of the training set to reserve for development, or None if no development set is needed.

        Raises:
            ValueError: If any of the split parameters are invalid.
        """
        if n_rows == 0:
            raise ValueError("Cannot split an empty dataset.")
        if test_size <= 0 or test_size >= 1:
            raise ValueError("test_size must be between 0 and 1.")
        if dev_size is not None and (dev_size <= 0 or dev_size >= 1):
            raise ValueError("dev_size must be between 0 and 1.")

    @staticmethod
    def _split_random_indices(
        n_rows: int, test_size: float, random_state: int
    ) -> dict[str, list[int]]:
        """Create a simple non-stratified random split.

        Args:
            n_rows: The total number of rows in the dataset.
            test_size: The proportion of the dataset to reserve for testing.
            random_state: The seed for the random number generator to ensure reproducibility.

        Returns:
            A dictionary containing the split row indices with keys "train" and "test".
        """
        rng = random.Random(random_state)
        indices = list(range(n_rows))
        rng.shuffle(indices)

        test_count = int(round(n_rows * test_size))
        if n_rows > 1 and test_count >= n_rows:
            test_count = n_rows - 1
        if n_rows > 1 and test_count <= 0:
            test_count = 1
        if n_rows <= 1:
            test_count = 0

        test_indices = set(indices[:test_count])
        train_indices = [idx for idx in indices if idx not in test_indices]
        return {"train": train_indices, "test": list(test_indices)}

    def split(
        self,
        test_size: float = 0.2,
        dev_size: float | None = None,
        random_state: int = 42,
        stratify: bool = True,
    ) -> dict[str, "ClassifierData"]:
        """Split a standardized dataset into train/test/dev partitions.

        Args:
            test_size: The proportion of the dataset to reserve for testing.
            dev_size: The proportion of the training set to reserve for development, or None if no development set is needed.
            random_state: The seed for the random number generator to ensure reproducibility.
            stratify: Whether to perform a stratified split based on the labels.

        Returns:
            A dictionary containing the split datasets with keys "train", "test", and optionally "dev".
        """
        n_rows = self.row_count()
        self._validate_split_parameters(n_rows, test_size, dev_size)

        if stratify and self.labels:
            split = self._split_indices_by_labels(
                test_size=test_size,
                dev_size=dev_size,
                random_state=random_state,
            )
        else:
            split = self._split_random_indices(n_rows, test_size, random_state)

        train_result = self.subset(split["train"])
        test_result = self.subset(split["test"])

        result: dict[str, ClassifierData] = {
            "train": train_result,
            "test": test_result,
        }

        if dev_size is not None and "dev" in split:
            result["dev"] = self.subset(split["dev"])

        return result

__init__(values: Any, labels: Sequence[Any] | None = None, *, docs: Any = None, titles: Sequence[Any] | None = None, matrix: Any = None, source: str = 'raw') -> None

Initialize the ClassifierData object.

Parameters:

Name	Type	Description	Default
values	Any	The main data values.	required
labels	Sequence[Any] | None	Optional sequence of labels corresponding to the data.	None
docs	Any	Optional sequence of document objects.	None
titles	Sequence[Any] | None	Optional sequence of titles for the data items.	None
matrix	Any	Optional matrix representation of the data.	None
source	str	A string indicating the source of the data.	'raw'

Source code in lexos/classification/classifier.py

def __init__(
    self,
    values: Any,
    labels: Sequence[Any] | None = None,
    *,
    docs: Any = None,
    titles: Sequence[Any] | None = None,
    matrix: Any = None,
    source: str = "raw",
) -> None:
    """Initialize the ClassifierData object.

    Args:
        values: The main data values.
        labels: Optional sequence of labels corresponding to the data.
        docs: Optional sequence of document objects.
        titles: Optional sequence of titles for the data items.
        matrix: Optional matrix representation of the data.
        source: A string indicating the source of the data.
    """
    super().__init__(
        values=values,
        labels=list(labels) if labels is not None else [],
        docs=docs,
        titles=list(titles) if titles is not None else None,
        matrix=matrix,
        source=source,
    )

as_texts() -> list[str]

Return the data as plain text strings when possible.

Returns:

Type	Description
list[str]	A list of plain text strings representing the data.

Source code in lexos/classification/classifier.py

def as_texts(self) -> list[str]:
    """Return the data as plain text strings when possible.

    Returns:
        A list of plain text strings representing the data.
    """
    if self.docs is not None:
        texts: list[str] = []
        for doc in self.docs:
            texts.append(self._as_text_for_doc(doc))
        return texts

    if isinstance(self.values, pd.DataFrame):
        return [str(item) for item in self.values.to_dict(orient="records")]

    if isinstance(self.values, (list, tuple)):
        texts: list[str] = []
        for item in self.values:
            if _is_spacy_doc(item):
                texts.append(" ".join(token.text for token in item))
            else:
                texts.append(str(item))
        return texts

    return [str(self.values)]

from_input(data: Any, labels: Sequence[Any] | None = None, titles: Sequence[Any] | None = None) -> ClassifierData classmethod

Normalize raw text, DataFrames, and Lexos DTM objects to a standard form.

Parameters:

Name	Type	Description	Default
data	Any	The input data to normalize. Can be raw text, a pandas DataFrame, a Lexos DTM object, or a matrix-like structure.	required
labels	Sequence[Any] | None	Optional sequence of labels to associate with the data. If not provided, labels will be inferred from the input data.	None
titles	Sequence[Any] | None	Optional sequence of titles aligned with the rows.	None

Returns:

Type	Description
ClassifierData	An instance of the ClassifierData class wrapping the normalized data.

Source code in lexos/classification/classifier.py

@classmethod
def from_input(
    cls,
    data: Any,
    labels: Sequence[Any] | None = None,
    titles: Sequence[Any] | None = None,
) -> "ClassifierData":
    """Normalize raw text, DataFrames, and Lexos DTM objects to a standard form.

    Args:
        data: The input data to normalize. Can be raw text, a pandas DataFrame, a Lexos DTM object, or a matrix-like structure.
        labels: Optional sequence of labels to associate with the data. If not provided, labels will be inferred from the input data.
        titles: Optional sequence of titles aligned with the rows.

    Returns:
        An instance of the ClassifierData class wrapping the normalized data.
    """
    if _is_dtm_like(data):
        return cls._from_dtm_input(data, labels, titles)
    if isinstance(data, pd.DataFrame):
        return cls._from_dataframe_input(data, labels, titles)
    if hasattr(data, "shape") and getattr(data, "ndim", None) == 2:
        return cls._from_matrix_input(data, labels, titles)
    if isinstance(data, (list, tuple)):
        if data and _is_spacy_doc(data[0]):
            resolved_labels, resolved_titles = cls._resolve_row_alignment(
                len(data),
                labels,
                titles,
                "Number of labels must match the number of data items.",
                "Title count must match the number of data items.",
            )
            return cls(
                values=list(data),
                labels=resolved_labels,
                titles=resolved_titles,
                docs=list(data),
                source="raw",
            )
        return cls._from_sequence_input(data, labels, titles)
    if data is None:
        raise ValueError("No data was supplied to the classifier.")

    return cls(
        values=data,
        labels=list(labels or []),
        titles=list(titles) if titles is not None else None,
        source="raw",
    )

row_count() -> int

Return the number of rows represented by the standardized input.

Returns:

Type	Description
int	The number of rows represented by the standardized input.

Source code in lexos/classification/classifier.py

def row_count(self) -> int:
    """Return the number of rows represented by the standardized input.

    Returns:
        The number of rows represented by the standardized input.
    """
    if self.matrix is not None:
        return self.matrix.shape[0]
    if self.docs is not None:
        return len(self.docs)
    if isinstance(self.values, pd.DataFrame):
        return len(self.values)
    if isinstance(self.values, (list, tuple)):
        return len(self.values)
    return len(self.labels)

split(test_size: float = 0.2, dev_size: float | None = None, random_state: int = 42, stratify: bool = True) -> dict[str, ClassifierData]

Split a standardized dataset into train/test/dev partitions.

Parameters:

Name	Type	Description	Default
test_size	float	The proportion of the dataset to reserve for testing.	0.2
dev_size	float | None	The proportion of the training set to reserve for development, or None if no development set is needed.	None
random_state	int	The seed for the random number generator to ensure reproducibility.	42
stratify	bool	Whether to perform a stratified split based on the labels.	True

Returns:

Type	Description
dict[str, ClassifierData]	A dictionary containing the split datasets with keys "train", "test", and optionally "dev".

Source code in lexos/classification/classifier.py

def split(
    self,
    test_size: float = 0.2,
    dev_size: float | None = None,
    random_state: int = 42,
    stratify: bool = True,
) -> dict[str, "ClassifierData"]:
    """Split a standardized dataset into train/test/dev partitions.

    Args:
        test_size: The proportion of the dataset to reserve for testing.
        dev_size: The proportion of the training set to reserve for development, or None if no development set is needed.
        random_state: The seed for the random number generator to ensure reproducibility.
        stratify: Whether to perform a stratified split based on the labels.

    Returns:
        A dictionary containing the split datasets with keys "train", "test", and optionally "dev".
    """
    n_rows = self.row_count()
    self._validate_split_parameters(n_rows, test_size, dev_size)

    if stratify and self.labels:
        split = self._split_indices_by_labels(
            test_size=test_size,
            dev_size=dev_size,
            random_state=random_state,
        )
    else:
        split = self._split_random_indices(n_rows, test_size, random_state)

    train_result = self.subset(split["train"])
    test_result = self.subset(split["test"])

    result: dict[str, ClassifierData] = {
        "train": train_result,
        "test": test_result,
    }

    if dev_size is not None and "dev" in split:
        result["dev"] = self.subset(split["dev"])

    return result

subset(indices: Sequence[int]) -> ClassifierData

Return a new data object containing only the selected row indices.

Parameters:

Name	Type	Description	Default
indices	Sequence[int]	A sequence of row indices to include in the subset.	required

Returns:

Type	Description
ClassifierData	A new ClassifierData object containing only the selected rows.

Source code in lexos/classification/classifier.py

def subset(self, indices: Sequence[int]) -> "ClassifierData":
    """Return a new data object containing only the selected row indices.

    Args:
        indices: A sequence of row indices to include in the subset.

    Returns:
        A new `ClassifierData` object containing only the selected rows.
    """
    idx_list = list(indices)
    selected_titles = (
        [self.titles[i] for i in idx_list] if self.titles is not None else None
    )
    if self.matrix is not None:
        return ClassifierData(
            values=self.matrix[idx_list],
            labels=[self.labels[i] for i in idx_list],
            docs=[self.docs[i] for i in idx_list]
            if self.docs is not None
            else None,
            titles=selected_titles,
            matrix=self.matrix[idx_list],
            source=self.source,
        )

    if isinstance(self.values, pd.DataFrame):
        data = self.values.iloc[idx_list]
        return ClassifierData(
            values=data,
            labels=[self.labels[i] for i in idx_list],
            titles=selected_titles,
            source=self.source,
        )

    sliced = [self.values[i] for i in idx_list]
    return ClassifierData(
        values=sliced,
        labels=[self.labels[i] for i in idx_list],
        titles=selected_titles,
        source=self.source,
    )

SklearnClassifierPipeline pydantic-model

Bases: BaseClassificationPipeline

A scikit-learn-based classification backend.

Fields:

• name (str)
• vectorizer (Any)
• estimator (Any)
• max_iter (int)
• multi_label_wrapper (Any)
• score_ranking (Literal['document', 'global'])

Source code in lexos/classification/sklearn_pipeline.py

class SklearnClassifierPipeline(BaseClassificationPipeline):
    """A scikit-learn-based classification backend."""

    vectorizer: Any = Field(
        default=None,
        description="Text vectorizer to transform raw text before fitting.",
    )
    estimator: Any = Field(
        default=None, description="Underlying scikit-learn estimator."
    )
    max_iter: int = Field(
        default=1000, description="Maximum number of iterations for iterative solvers."
    )
    multi_label_wrapper: Any = Field(
        default=None,
        description=(
            "Optional factory that wraps a base estimator for multi-label tasks. "
            "Receives a base estimator and returns a multi-label-compatible estimator."
        ),
    )
    score_ranking: Literal["document", "global"] = Field(
        default="document",
        description=(
            "How to rank labels when selecting multi-label outputs: 'document' "
            "sorts each document's scores separately, while 'global' ranks labels "
            "according to their score across the whole prediction set."
        ),
    )
    _target_label_count: int = PrivateAttr(default=1)
    _global_label_rank: dict[str, int] = PrivateAttr(default_factory=dict)

    def __init__(self, **data):
        """Initialize the SklearnClassifierPipeline with the specified settings."""
        super().__init__(**data)
        if self.vectorizer is None:
            if TfidfVectorizer is None:
                raise ImportError(
                    "scikit-learn is required for SklearnClassifierPipeline."
                )
            self.vectorizer = TfidfVectorizer()
        if self.estimator is None:
            if LogisticRegression is None:
                raise ImportError(
                    "scikit-learn is required for SklearnClassifierPipeline."
                )
            self.estimator = LogisticRegression(max_iter=self.max_iter)

    def save(self, path: str | Path) -> None:
        """Save the fitted sklearn pipeline together with its configuration."""
        path = Path(path)
        path.parent.mkdir(parents=True, exist_ok=True)
        payload = {
            "config": self.model_dump(mode="python"),
            "vectorizer": self.vectorizer,
            "estimator": self.estimator,
            "label_binarizer": getattr(self, "_label_binarizer", None),
            "target_label_count": getattr(self, "_target_label_count", 1),
            "global_label_rank": getattr(self, "_global_label_rank", {}),
        }
        joblib.dump(payload, path)

    @classmethod
    def load(cls, path: str | Path) -> "SklearnClassifierPipeline":
        """Load a saved sklearn pipeline and restore its fitted state and config."""
        payload = joblib.load(Path(path))
        pipeline = cls(**payload["config"])
        pipeline.vectorizer = payload["vectorizer"]
        pipeline.estimator = payload["estimator"]
        pipeline._label_binarizer = payload.get("label_binarizer")
        pipeline._target_label_count = payload.get("target_label_count", 1)
        pipeline._global_label_rank = payload.get("global_label_rank", {})
        return pipeline

    def _coerce_label_list(self, value: Any) -> list[str]:
        """Normalize a single label or list-valued document target to a flat label list.

        Args:
            value: The raw label or list of labels for a single document.

        Returns:
            A flat list of string labels.
        """
        if value is None:
            return []
        if isinstance(value, str):
            return [value]
        if isinstance(value, (list, tuple, set)):
            labels: list[str] = []
            for item in value:
                labels.extend(self._coerce_label_list(item))
            return labels
        return [str(value)]

    def _is_multi_label(self, labels: Sequence[Any]) -> bool:
        """Return True when any document target contains more than one label.

        Args:
            labels: A sequence of raw label targets for each document.

        Returns:
            True if any document has more than one label, False otherwise.
        """
        return any(len(self._coerce_label_list(label)) > 1 for label in labels)

    def _prepare_targets(self, labels: Sequence[Any]) -> tuple[list[str], Any | None]:
        """Convert single-label or multi-label targets into a training-ready representation.

        Args:
            labels: A sequence of raw label targets for each document.

        Returns:
            A tuple containing the list of unique labels and the encoded target matrix (or None for single-label).
        """
        if not labels:
            return [], None
        if not self._is_multi_label(labels):
            single_labels: list[str] = []
            for document_label in labels:
                single_labels.extend(self._coerce_label_list(document_label))
            return [str(label) for label in single_labels], None

        flattened = [self._coerce_label_list(label) for label in labels]
        mlb = MultiLabelBinarizer()
        encoded = mlb.fit_transform(flattened)
        return [str(value) for value in mlb.classes_], encoded

    @property
    def model(self) -> Any:
        """Return the underlying scikit-learn estimator.

        Returns:
            The underlying scikit-learn estimator object.
        """
        return self.estimator

    def _prepare_label_lists(self, labels: Sequence[Any]) -> list[list[str]]:
        """Normalize raw label targets to a list of per-document label sets.

        Args:
            labels: A sequence of raw label targets for each document.

        Returns:
            A list of per-document label lists.
        """
        return [self._coerce_label_list(label) for label in labels]

    def _get_estimator_for(self, is_multi: bool) -> Any:
        """Create the underlying estimator for the given label scheme.

        Args:
            is_multi: A boolean indicating whether the label scheme is multi-label.

        Returns:
            The underlying scikit-learn estimator object configured for the label scheme.
        """
        if not is_multi:
            return self.estimator

        base_estimator = LogisticRegression(max_iter=self.max_iter)
        if self.multi_label_wrapper is not None:
            return self.multi_label_wrapper(base_estimator)
        return OneVsRestClassifier(base_estimator)

    def _fit_label_matrix(
        self, matrix: Any, label_list: list[list[str]], is_multi: bool
    ) -> None:
        """Fit a matrix-like feature representation against single- or multi-label targets.

        Args:
            matrix: A matrix-like feature representation of the input data.
            label_list: A list of per-document label lists.
            is_multi: A boolean indicating whether the label scheme is multi-label.

        Returns:
            None
        """
        if is_multi:
            self._target_label_count = max(
                (len(item) for item in label_list), default=1
            )
            self.estimator = self._get_estimator_for(is_multi=True)
            mlb = MultiLabelBinarizer()
            transformed_targets = mlb.fit_transform(label_list)
            self.estimator.fit(matrix, transformed_targets)
            self._label_binarizer = mlb
            return

        self._target_label_count = 1
        self.estimator.fit(
            matrix, [str(item[0]) if item else "" for item in label_list]
        )

    def _fit_text_matrix(
        self, texts: Sequence[str], label_list: list[list[str]], is_multi: bool
    ) -> None:
        """Vectorize texts and fit the model against the prepared labels.

        Args:
            texts: A sequence of raw text documents.
            label_list: A list of per-document label lists.
            is_multi: A boolean indicating whether the label scheme is multi-label.

        Returns:
            None
        """
        if len(texts) != len(label_list):
            raise ValueError("Number of texts must match the number of labels.")

        transformed = self.vectorizer.fit_transform(texts)
        self._fit_label_matrix(transformed, label_list, is_multi)

    @staticmethod
    def _is_spacy_doc(doc: Any) -> bool:
        """Return True when `doc` is a spaCy Doc-like object."""
        return (
            doc is not None
            and type(doc).__module__.startswith("spacy.")
            and hasattr(doc, "vocab")
            and hasattr(doc, "__iter__")
        )

    def _coerce_text_for_doc(self, doc: Any) -> str:
        """Convert a document-like item to text while preserving spaCy tokenization."""
        if self._is_spacy_doc(doc):
            return " ".join(token.text for token in doc)
        return str(doc)

    def _coerce_texts(self, data: Any) -> list[str]:
        """Normalize raw data into strings while preserving spaCy Doc tokenization."""
        if isinstance(data, ClassifierData):
            if data.docs is not None:
                return [self._coerce_text_for_doc(doc) for doc in data.docs]
            return data.as_texts()

        if hasattr(data, "docs") and getattr(data, "docs", None) is not None:
            return [self._coerce_text_for_doc(doc) for doc in data.docs]

        if isinstance(data, (list, tuple)):
            return [self._coerce_text_for_doc(item) for item in data]

        return [str(item) for item in data]

    def fit(self, data: Any, labels: Sequence[Any]) -> "SklearnClassifierPipeline":
        """Fit a scikit-learn classifier on the provided data.

        Args:
            data: Raw text, a ClassifierData object, a Lexos DTM, or another matrix-like object.
            labels: A sequence of label strings or per-document label collections.

        Returns:
            The fitted `SklearnClassifierPipeline` instance.
        """
        if TfidfVectorizer is None or LogisticRegression is None:
            raise ImportError("scikit-learn is required for SklearnClassifierPipeline.")

        label_list = self._prepare_label_lists(labels)
        is_multi = any(len(item) > 1 for item in label_list)

        if isinstance(data, ClassifierData):
            matrix = data.matrix
            if matrix is not None:
                self._fit_label_matrix(matrix, label_list, is_multi)
                return self

            texts = self._coerce_texts(data)
            self._fit_text_matrix(texts, label_list, is_multi)
            return self

        if hasattr(data, "doc_term_matrix") and hasattr(data, "vectorizer"):
            matrix = getattr(data, "doc_term_matrix")
            if matrix is None:
                raise ValueError("DTM input is missing a document-term matrix.")
            self._fit_label_matrix(matrix, label_list, is_multi)
            return self

        if hasattr(data, "shape") and getattr(data, "ndim", None) == 2:
            self._fit_label_matrix(data, label_list, is_multi)
            return self

        texts = self._coerce_texts(data)
        self._fit_text_matrix(texts, label_list, is_multi)
        return self

    def _rank_score_row(self, score_row: dict[str, float]) -> dict[str, float]:
        """Rank a single score row according to the configured scoring policy."""
        if not score_row:
            return {}

        if self.score_ranking == "document":
            return {
                label: score
                for label, score in sorted(
                    score_row.items(), key=lambda item: item[1], reverse=True
                )
            }

        global_rank = getattr(self, "_global_label_rank", {})
        return {
            label: score
            for label, score in sorted(
                score_row.items(),
                key=lambda item: (
                    item[1],
                    -global_rank.get(item[0], 0),
                ),
                reverse=True,
            )
        }

    def _prepare_prediction_matrix(self, data: Any) -> Any:
        """Normalize an input payload into the matrix structure expected by the estimator."""
        if self.estimator is None:
            raise ValueError("The pipeline must be fitted before calling predict().")

        if isinstance(data, ClassifierData):
            if data.matrix is not None:
                return data.matrix
            if self.vectorizer is None:
                raise ValueError(
                    "The pipeline must be fitted before calling predict()."
                )
            return self.vectorizer.transform(data.as_texts())

        if hasattr(data, "doc_term_matrix") and hasattr(data, "vectorizer"):
            matrix = getattr(data, "doc_term_matrix")
            if matrix is None:
                raise ValueError("DTM input is missing a document-term matrix.")
            return matrix

        if hasattr(data, "shape") and getattr(data, "ndim", None) == 2:
            return data

        if self.vectorizer is None:
            raise ValueError("The pipeline must be fitted before calling predict().")
        return self.vectorizer.transform(self._coerce_texts(data))

    def _build_global_label_rank(
        self, score_rows: Sequence[dict[str, float]]
    ) -> dict[str, int]:
        """Compute a corpus-level rank for labels from a sequence of score rows."""
        if not score_rows:
            return {}

        label_scores: dict[str, list[float]] = {}
        for row in score_rows:
            for label, score in row.items():
                label_scores.setdefault(label, []).append(float(score))

        return {
            label: rank
            for rank, label in enumerate(
                sorted(
                    label_scores,
                    key=lambda label: (
                        sum(label_scores[label]) / len(label_scores[label]),
                        label,
                    ),
                    reverse=True,
                )
            )
        }

    def _score_rows_for_matrix(self, matrix: Any) -> list[dict[str, float]]:
        """Return raw label scores for each row in the given matrix."""
        label_binarizer = getattr(self, "_label_binarizer", None)
        if label_binarizer is not None:
            classes = list(label_binarizer.classes_)
            if hasattr(self.estimator, "predict_proba"):
                scores = self.estimator.predict_proba(matrix)
            elif hasattr(self.estimator, "decision_function"):
                scores = self.estimator.decision_function(matrix)
            else:
                return []
            return [
                {str(classes[idx]): float(score) for idx, score in enumerate(row)}
                for row in scores
            ]

        if hasattr(self.estimator, "predict_proba"):
            scores = self.estimator.predict_proba(matrix)
            return [
                {str(index): float(score) for index, score in enumerate(row)}
                for row in scores
            ]

        raise NotImplementedError(
            f"{type(self.estimator).__name__} does not implement predict_scores()."
        )

    def _select_top_labels(self, score_row: dict[str, float]) -> list[str]:
        """Return the top labels for a single document according to the configured ranking."""
        ranked = self._rank_score_row(score_row)
        top_count = max(1, self._target_label_count)
        return list(ranked.keys())[:top_count]

    def predict(self, data: Any) -> list[str]:
        """Predict labels for a sequence of texts, a ClassifierData object, or a Lexos DTM.

        Args:
            data: Raw text, a ClassifierData object, a Lexos DTM, or another matrix-like object.

        Returns:
            A list of predicted label strings.
        """
        if self.estimator is None:
            raise ValueError("The pipeline must be fitted before calling predict().")

        matrix_to_predict = self._prepare_prediction_matrix(data)
        predictions = self.estimator.predict(matrix_to_predict)

        if getattr(self, "_label_binarizer", None) is None:
            return [str(value) for value in predictions]

        score_rows = self._score_rows_for_matrix(matrix_to_predict)
        if score_rows:
            if self.score_ranking == "global":
                self._global_label_rank = self._build_global_label_rank(score_rows)
            else:
                self._global_label_rank = {}
            return [self._select_top_labels(row) for row in score_rows]

        decoded = self._label_binarizer.inverse_transform(predictions)
        return [[str(label) for label in labels] for labels in decoded]

    def predict_scores(self, data: Any) -> list[dict[str, float]]:
        """Return the per-document confidence scores for each label.

        Args:
            data: Raw text, a ClassifierData object, a Lexos DTM, or another matrix-like object.

        Returns:
            A list of dictionaries containing each label's score for each document.
        """
        if self.estimator is None:
            raise ValueError(
                "The pipeline must be fitted before calling predict_scores()."
            )

        matrix_to_predict = self._prepare_prediction_matrix(data)
        raw_rows = self._score_rows_for_matrix(matrix_to_predict)

        if self.score_ranking == "global" and raw_rows:
            self._global_label_rank = self._build_global_label_rank(raw_rows)
        else:
            self._global_label_rank = {}

        return [self._rank_score_row(row) for row in raw_rows]

    def evaluate(self, data: Any, labels: Sequence[Any]) -> dict[str, float]:
        """Evaluate using standard accuracy.

        Args:
            data: The input text data to evaluate.
            labels: A sequence of true label strings or per-document label lists.

        Returns:
            A dictionary containing the accuracy of the predictions.
        """
        if accuracy_score is None:
            raise ImportError("scikit-learn is required to compute evaluation metrics.")
        predictions = self.predict(data)
        if getattr(self, "_label_binarizer", None) is not None:
            gold = [set(self._coerce_label_list(label)) for label in labels]
            pred = [
                set(item) if isinstance(item, list) else set() for item in predictions
            ]
            matches = sum(1 for a, b in zip(gold, pred) if a == b)
            return {"accuracy": float(matches / len(labels))}
        return {"accuracy": float(accuracy_score(list(labels), predictions))}

estimator: Any = None pydantic-field

Underlying scikit-learn estimator.

max_iter: int = 1000 pydantic-field

Maximum number of iterations for iterative solvers.

model: Any property

Return the underlying scikit-learn estimator.

Returns:

Type	Description
Any	The underlying scikit-learn estimator object.

multi_label_wrapper: Any = None pydantic-field

Optional factory that wraps a base estimator for multi-label tasks. Receives a base estimator and returns a multi-label-compatible estimator.

name: str = 'classifier' pydantic-field

Human-readable pipeline name.

score_ranking: Literal['document', 'global'] = 'document' pydantic-field

How to rank labels when selecting multi-label outputs: 'document' sorts each document's scores separately, while 'global' ranks labels according to their score across the whole prediction set.

vectorizer: Any = None pydantic-field

Text vectorizer to transform raw text before fitting.

__call__(data: Any) -> Sequence[str]

Convenience wrapper for predicting on a single data payload.

Source code in lexos/classification/classifier.py

def __call__(self, data: Any) -> Sequence[str]:
    """Convenience wrapper for predicting on a single data payload."""
    return self.predict(data)

__init__(**data)

Initialize the SklearnClassifierPipeline with the specified settings.

Source code in lexos/classification/sklearn_pipeline.py

def __init__(self, **data):
    """Initialize the SklearnClassifierPipeline with the specified settings."""
    super().__init__(**data)
    if self.vectorizer is None:
        if TfidfVectorizer is None:
            raise ImportError(
                "scikit-learn is required for SklearnClassifierPipeline."
            )
        self.vectorizer = TfidfVectorizer()
    if self.estimator is None:
        if LogisticRegression is None:
            raise ImportError(
                "scikit-learn is required for SklearnClassifierPipeline."
            )
        self.estimator = LogisticRegression(max_iter=self.max_iter)

evaluate(data: Any, labels: Sequence[Any]) -> dict[str, float]

Evaluate using standard accuracy.

Parameters:

Name	Type	Description	Default
data	Any	The input text data to evaluate.	required
labels	Sequence[Any]	A sequence of true label strings or per-document label lists.	required

Returns:

Type	Description
dict[str, float]	A dictionary containing the accuracy of the predictions.

Source code in lexos/classification/sklearn_pipeline.py

def evaluate(self, data: Any, labels: Sequence[Any]) -> dict[str, float]:
    """Evaluate using standard accuracy.

    Args:
        data: The input text data to evaluate.
        labels: A sequence of true label strings or per-document label lists.

    Returns:
        A dictionary containing the accuracy of the predictions.
    """
    if accuracy_score is None:
        raise ImportError("scikit-learn is required to compute evaluation metrics.")
    predictions = self.predict(data)
    if getattr(self, "_label_binarizer", None) is not None:
        gold = [set(self._coerce_label_list(label)) for label in labels]
        pred = [
            set(item) if isinstance(item, list) else set() for item in predictions
        ]
        matches = sum(1 for a, b in zip(gold, pred) if a == b)
        return {"accuracy": float(matches / len(labels))}
    return {"accuracy": float(accuracy_score(list(labels), predictions))}

fit(data: Any, labels: Sequence[Any]) -> SklearnClassifierPipeline

Fit a scikit-learn classifier on the provided data.

Parameters:

Name	Type	Description	Default
data	Any	Raw text, a ClassifierData object, a Lexos DTM, or another matrix-like object.	required
labels	Sequence[Any]	A sequence of label strings or per-document label collections.	required

Returns:

Type	Description
SklearnClassifierPipeline	The fitted SklearnClassifierPipeline instance.

Source code in lexos/classification/sklearn_pipeline.py

def fit(self, data: Any, labels: Sequence[Any]) -> "SklearnClassifierPipeline":
    """Fit a scikit-learn classifier on the provided data.

    Args:
        data: Raw text, a ClassifierData object, a Lexos DTM, or another matrix-like object.
        labels: A sequence of label strings or per-document label collections.

    Returns:
        The fitted `SklearnClassifierPipeline` instance.
    """
    if TfidfVectorizer is None or LogisticRegression is None:
        raise ImportError("scikit-learn is required for SklearnClassifierPipeline.")

    label_list = self._prepare_label_lists(labels)
    is_multi = any(len(item) > 1 for item in label_list)

    if isinstance(data, ClassifierData):
        matrix = data.matrix
        if matrix is not None:
            self._fit_label_matrix(matrix, label_list, is_multi)
            return self

        texts = self._coerce_texts(data)
        self._fit_text_matrix(texts, label_list, is_multi)
        return self

    if hasattr(data, "doc_term_matrix") and hasattr(data, "vectorizer"):
        matrix = getattr(data, "doc_term_matrix")
        if matrix is None:
            raise ValueError("DTM input is missing a document-term matrix.")
        self._fit_label_matrix(matrix, label_list, is_multi)
        return self

    if hasattr(data, "shape") and getattr(data, "ndim", None) == 2:
        self._fit_label_matrix(data, label_list, is_multi)
        return self

    texts = self._coerce_texts(data)
    self._fit_text_matrix(texts, label_list, is_multi)
    return self

load(path: str | Path) -> SklearnClassifierPipeline classmethod

Load a saved sklearn pipeline and restore its fitted state and config.

Source code in lexos/classification/sklearn_pipeline.py

@classmethod
def load(cls, path: str | Path) -> "SklearnClassifierPipeline":
    """Load a saved sklearn pipeline and restore its fitted state and config."""
    payload = joblib.load(Path(path))
    pipeline = cls(**payload["config"])
    pipeline.vectorizer = payload["vectorizer"]
    pipeline.estimator = payload["estimator"]
    pipeline._label_binarizer = payload.get("label_binarizer")
    pipeline._target_label_count = payload.get("target_label_count", 1)
    pipeline._global_label_rank = payload.get("global_label_rank", {})
    return pipeline

predict(data: Any) -> list[str]

Predict labels for a sequence of texts, a ClassifierData object, or a Lexos DTM.

Parameters:

Name	Type	Description	Default
data	Any	Raw text, a ClassifierData object, a Lexos DTM, or another matrix-like object.	required

Returns:

Type	Description
list[str]	A list of predicted label strings.

Source code in lexos/classification/sklearn_pipeline.py

def predict(self, data: Any) -> list[str]:
    """Predict labels for a sequence of texts, a ClassifierData object, or a Lexos DTM.

    Args:
        data: Raw text, a ClassifierData object, a Lexos DTM, or another matrix-like object.

    Returns:
        A list of predicted label strings.
    """
    if self.estimator is None:
        raise ValueError("The pipeline must be fitted before calling predict().")

    matrix_to_predict = self._prepare_prediction_matrix(data)
    predictions = self.estimator.predict(matrix_to_predict)

    if getattr(self, "_label_binarizer", None) is None:
        return [str(value) for value in predictions]

    score_rows = self._score_rows_for_matrix(matrix_to_predict)
    if score_rows:
        if self.score_ranking == "global":
            self._global_label_rank = self._build_global_label_rank(score_rows)
        else:
            self._global_label_rank = {}
        return [self._select_top_labels(row) for row in score_rows]

    decoded = self._label_binarizer.inverse_transform(predictions)
    return [[str(label) for label in labels] for labels in decoded]

predict_scores(data: Any) -> list[dict[str, float]]

Return the per-document confidence scores for each label.

Parameters:

Name	Type	Description	Default
data	Any	Raw text, a ClassifierData object, a Lexos DTM, or another matrix-like object.	required

Returns:

Type	Description
list[dict[str, float]]	A list of dictionaries containing each label's score for each document.

Source code in lexos/classification/sklearn_pipeline.py

def predict_scores(self, data: Any) -> list[dict[str, float]]:
    """Return the per-document confidence scores for each label.

    Args:
        data: Raw text, a ClassifierData object, a Lexos DTM, or another matrix-like object.

    Returns:
        A list of dictionaries containing each label's score for each document.
    """
    if self.estimator is None:
        raise ValueError(
            "The pipeline must be fitted before calling predict_scores()."
        )

    matrix_to_predict = self._prepare_prediction_matrix(data)
    raw_rows = self._score_rows_for_matrix(matrix_to_predict)

    if self.score_ranking == "global" and raw_rows:
        self._global_label_rank = self._build_global_label_rank(raw_rows)
    else:
        self._global_label_rank = {}

    return [self._rank_score_row(row) for row in raw_rows]

save(path: str | Path) -> None

Save the fitted sklearn pipeline together with its configuration.

Source code in lexos/classification/sklearn_pipeline.py

def save(self, path: str | Path) -> None:
    """Save the fitted sklearn pipeline together with its configuration."""
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    payload = {
        "config": self.model_dump(mode="python"),
        "vectorizer": self.vectorizer,
        "estimator": self.estimator,
        "label_binarizer": getattr(self, "_label_binarizer", None),
        "target_label_count": getattr(self, "_target_label_count", 1),
        "global_label_rank": getattr(self, "_global_label_rank", {}),
    }
    joblib.dump(payload, path)

SpaCyTextCategorizerPipeline pydantic-model

Bases: BaseClassificationPipeline

A spaCy TextCategorizer wrapper with a Lexos-friendly API.

Fields:

• name (str)
• language (str)
• nlp (Language | None)
• exclusive_classes (bool)
• architecture (str | dict[str, Any])
• epochs (int)
• score_ranking (Literal['document', 'global'])

Validators:

• _validate_and_resolve_settings

Source code in lexos/classification/spacy_pipeline.py

class SpaCyTextCategorizerPipeline(BaseClassificationPipeline):
    """A spaCy `TextCategorizer` wrapper with a Lexos-friendly API."""

    language: str = Field(default="en", description="SpaCy language shortcut to use.")
    nlp: Language | None = Field(
        default=None, description="Underlying spaCy language object."
    )
    exclusive_classes: bool = Field(
        default=True, description="Whether classes are mutually exclusive."
    )
    architecture: str | dict[str, Any] = Field(
        default="bow", description="Text categorizer architecture or custom config."
    )
    epochs: int = Field(default=10, description="Number of training epochs.")
    score_ranking: Literal["document", "global"] = Field(
        default="document",
        description=(
            "How to rank labels when selecting a multi-label prediction: "
            "'document' ranks within each document, 'global' ranks using the "
            "global score ordering across all predicted documents."
        ),
    )
    _labels: list[str] = PrivateAttr(default_factory=list)
    _resolved_pipe_name: str = PrivateAttr(default="textcat")
    _resolved_architecture: str = PrivateAttr(default="bow")
    _target_label_count: int = PrivateAttr(default=1)

    @model_validator(mode="after")
    def _validate_and_resolve_settings(self):
        """Resolve the configured textcat behavior into a valid spaCy component config.

        Args:
            mode (str): The validation mode, typically "after".
        """
        if isinstance(self.architecture, dict):
            custom_cfg = dict(self.architecture)
            if "model" in custom_cfg:
                self._resolved_architecture = custom_cfg["model"]
            else:
                self._resolved_architecture = custom_cfg
            if "exclusive_classes" in self._resolved_architecture:
                self.exclusive_classes = bool(
                    self._resolved_architecture["exclusive_classes"]
                )
            self._resolved_pipe_name = (
                "textcat_multilabel" if not self.exclusive_classes else "textcat"
            )
            return self

        if self.architecture not in {"bow", "cnn", "ensemble"}:
            raise ValueError(
                "Unsupported spaCy architecture. Expected one of: bow, cnn, ensemble."
            )

        self._resolved_architecture = self.architecture
        self._resolved_pipe_name = (
            "textcat_multilabel" if not self.exclusive_classes else "textcat"
        )
        return self

    def __init__(self, **data):
        """Initialize the SpaCyTextCategorizerPipeline with the specified settings."""
        super().__init__(**data)
        if self.nlp is None:
            self.nlp = spacy.blank(self.language)

        pipe_name = self._resolved_pipe_name
        if pipe_name not in self.nlp.pipe_names:
            cfg = self._build_pipe_config()
            self.nlp.add_pipe(pipe_name, config=cfg)

    def save(self, path: str | Path) -> None:
        """Save the trained spaCy pipeline and its configuration to disk."""
        target = Path(path)
        target.mkdir(parents=True, exist_ok=True)
        config = self.model_dump(mode="python", exclude={"nlp"})
        config["_labels"] = list(self._labels)
        config["_target_label_count"] = int(getattr(self, "_target_label_count", 1))
        config["_global_label_rank"] = dict(getattr(self, "_global_label_rank", {}))
        config_path = target / "pipeline_config.json"
        config_path.write_text(json.dumps(config, indent=2))
        self.nlp.to_disk(str(target / "model"))

    @classmethod
    def load(cls, path: str | Path) -> "SpaCyTextCategorizerPipeline":
        """Load a saved spaCy pipeline instance and restore its configuration."""
        target = Path(path)
        config = json.loads((target / "pipeline_config.json").read_text())
        pipeline = cls(**{k: v for k, v in config.items() if not k.startswith("_")})
        pipeline.nlp = spacy.load(str(target / "model"))
        pipeline._labels = list(config.get("_labels", []))
        pipeline._target_label_count = int(config.get("_target_label_count", 1))
        pipeline._global_label_rank = dict(config.get("_global_label_rank", {}))
        pipeline._resolved_pipe_name = (
            "textcat_multilabel" if not pipeline.exclusive_classes else "textcat"
        )
        return pipeline

    def _build_pipe_config(self) -> dict[str, Any]:
        """Resolve the configured architecture into the model config spaCy expects.

        Returns:
            dict[str, Any]: The resolved spaCy pipe configuration.
        """
        if isinstance(self._resolved_architecture, dict):
            custom_cfg = dict(self._resolved_architecture)
            if "@architectures" not in custom_cfg and "model" in custom_cfg:
                custom_cfg = dict(custom_cfg["model"])
            if "exclusive_classes" not in custom_cfg:
                custom_cfg["exclusive_classes"] = self.exclusive_classes
            if "no_output_layer" not in custom_cfg:
                custom_cfg["no_output_layer"] = False
            return {"model": custom_cfg}

        architecture = self._resolved_architecture
        base = {
            "@architectures": "spacy.TextCatBOW.v3",
            "ngram_size": 1,
            "length": 262144,
            "no_output_layer": False,
        }
        if architecture == "bow":
            base["exclusive_classes"] = self.exclusive_classes
            return {"model": base}

        if architecture == "cnn":
            base = {
                "@architectures": "spacy.TextCatReduce.v1",
                "exclusive_classes": self.exclusive_classes,
                "use_reduce_first": False,
                "use_reduce_last": False,
                "use_reduce_max": False,
                "use_reduce_mean": True,
                "tok2vec": {
                    "@architectures": "spacy.HashEmbedCNN.v2",
                    "pretrained_vectors": None,
                    "width": 96,
                    "depth": 4,
                    "embed_size": 2000,
                    "window_size": 1,
                    "maxout_pieces": 3,
                    "subword_features": True,
                },
            }
            return {"model": base}

        if architecture == "ensemble":
            return {
                "model": {
                    "@architectures": "spacy.TextCatEnsemble.v2",
                    "tok2vec": {
                        "@architectures": "spacy.Tok2Vec.v2",
                        "embed": {
                            "@architectures": "spacy.MultiHashEmbed.v2",
                            "width": 64,
                            "rows": [2000, 2000, 500, 1000, 500],
                            "attrs": ["NORM", "LOWER", "PREFIX", "SUFFIX", "SHAPE"],
                            "include_static_vectors": False,
                        },
                        "encode": {
                            "@architectures": "spacy.MaxoutWindowEncoder.v2",
                            "width": 64,
                            "window_size": 1,
                            "maxout_pieces": 3,
                            "depth": 2,
                        },
                    },
                    "linear_model": {
                        "@architectures": "spacy.TextCatBOW.v3",
                        "exclusive_classes": self.exclusive_classes,
                        "length": 262144,
                        "ngram_size": 1,
                        "no_output_layer": False,
                    },
                }
            }

        raise ValueError(f"Unsupported spaCy architecture: {architecture}")

    @property
    def model(self) -> Language:
        """Return the underlying spaCy language model.

        Returns:
            The underlying spaCy `Language` object representing the text categorizer model.
        """
        return self.nlp

    def _coerce_label_list(self, value: Any) -> list[str]:
        """Normalize a single label or a per-document label collection to a list of strings.

        Args:
            value: A single label, a collection of labels, or None.

        Returns:
            A list of strings representing the normalized labels.
        """
        if value is None:
            return []
        if isinstance(value, str):
            return [value]
        if isinstance(value, (list, tuple, set)):
            labels: list[str] = []
            for item in value:
                labels.extend(self._coerce_label_list(item))
            return labels
        return [str(value)]

    def _build_score_map(
        self, categories: Sequence[str], labels: Any
    ) -> dict[str, float]:
        """Build a spaCy `cats` score map from a document's active label or label set.

        Args:
            categories: A sequence of all possible category labels.
            labels: The active label or collection of labels for the current document.

        Returns:
            A dictionary mapping each category to a float score (1.0 for active labels, 0.0 otherwise).
        """
        score_map = {category: 0.0 for category in categories}
        for label in self._coerce_label_list(labels):
            score_map[str(label)] = 1.0
        return score_map

    def _get_textcat_pipe(self):
        """Return the active spaCy text categorizer component for this pipeline."""
        pipe_name = self._resolved_pipe_name
        if pipe_name not in self.nlp.pipe_names:
            raise KeyError(
                f"[E001] No component '{pipe_name}' found in pipeline. "
                f"Available names: {list(self.nlp.pipe_names)}"
            )
        return self.nlp.get_pipe(pipe_name)

    def _switch_to_multi_label_mode(self) -> None:
        """Switch the underlying spaCy text categorizer to multi-label mode."""
        if self.exclusive_classes is False:
            self._resolved_pipe_name = "textcat_multilabel"
            return

        self.exclusive_classes = False
        current_pipe_name = self._resolved_pipe_name
        target_pipe_name = "textcat_multilabel"

        if (
            current_pipe_name != target_pipe_name
            and current_pipe_name in self.nlp.pipe_names
        ):
            self.nlp.remove_pipe(current_pipe_name)
        self._resolved_pipe_name = target_pipe_name
        if target_pipe_name not in self.nlp.pipe_names:
            self.nlp.add_pipe(target_pipe_name, config=self._build_pipe_config())

    def _ensure_labels(self, labels: Sequence[Any]) -> list[str]:
        """Ensure that the specified labels are added to the text categorizer.

        Args:
            labels: A sequence of label strings or per-document label collections.

        Returns:
            A sorted list of unique labels that are now present in the text categorizer.
        """
        flattened = []
        for item in labels:
            flattened.extend(self._coerce_label_list(item))
        unique_labels = sorted({str(label) for label in flattened})
        textcat_pipe = self._get_textcat_pipe()
        for label in unique_labels:
            if label not in textcat_pipe.labels:
                textcat_pipe.add_label(label)
        self._labels = unique_labels
        return unique_labels

    @staticmethod
    def _is_spacy_doc(doc: Any) -> bool:
        """Return True when `doc` is a spaCy Doc-like object."""
        return (
            doc is not None
            and type(doc).__module__.startswith("spacy.")
            and hasattr(doc, "vocab")
            and hasattr(doc, "__iter__")
        )

    @staticmethod
    def _coerce_doc_to_text(doc: Any) -> str:
        """Convert a document-like object to a text string while preserving tokenization."""
        if doc is None:
            return ""
        if SpaCyTextCategorizerPipeline._is_spacy_doc(doc):
            return " ".join(token.text for token in doc)
        if hasattr(doc, "text") and not SpaCyTextCategorizerPipeline._is_spacy_doc(doc):
            return str(doc.text)
        if isinstance(doc, (list, tuple, set)):
            return " ".join(str(item) for item in doc)
        return str(doc)

    def _coerce_texts_from_sequence(self, docs: Sequence[Any]) -> list[str]:
        """Normalize a sequence of doc-like items into plain text strings."""
        return [self._coerce_doc_to_text(doc) for doc in docs]

    def _coerce_texts_from_input(self, data: Any) -> list[str]:
        """Convert raw text, a standardized dataset, or Lexos docs into spaCy text strings.

        Args:
            data: The input data, which can be raw text, a standardized dataset, or Lexos docs.

        Returns:
            A list of spaCy-compatible text strings extracted from the input data.
        """
        if isinstance(data, ClassifierData):
            if data.matrix is not None and data.docs is None:
                raise ValueError(
                    "SpaCy text categorization requires tokenized document text, not just a sparse DTM matrix."
                )
            return data.as_texts()

        if hasattr(data, "docs") and getattr(data, "docs", None) is not None:
            return self._coerce_texts_from_sequence(data.docs)

        if hasattr(data, "doc_term_matrix") and hasattr(data, "vectorizer"):
            docs = getattr(data, "docs", None)
            if docs is None:
                raise ValueError(
                    "SpaCy text categorization requires tokenized document text, not just a sparse DTM matrix."
                )
            return self._coerce_texts_from_sequence(docs)

        return [str(item) for item in data]

    def fit(self, data: Any, labels: Sequence[Any]) -> "SpaCyTextCategorizerPipeline":
        """Fit a spaCy text categorizer on raw text or tokenized Lexos docs.

        Args:
            data: The input data, which can be raw text, a standardized dataset, or Lexos docs.
            labels: The corresponding labels for each document in the input data.

        Returns:
            The fitted SpaCyTextCategorizerPipeline instance.
        """
        texts = self._coerce_texts_from_input(data)
        label_values = list(labels)

        if len(texts) != len(label_values):
            raise ValueError("Number of texts must match the number of labels.")

        normalized_labels = [self._coerce_label_list(label) for label in label_values]
        if self.exclusive_classes and any(len(item) > 1 for item in normalized_labels):
            self._switch_to_multi_label_mode()

        if self.exclusive_classes:
            for i, item in enumerate(normalized_labels):
                if len(item) != 1:
                    raise ValueError(
                        "exclusive_classes=True requires exactly one label per document; "
                        f"found {item!r} for row {i}."
                    )

        categories = self._ensure_labels(normalized_labels)
        if not categories:
            raise ValueError("At least one label is required before training.")

        self._target_label_count = max(
            (len(item) for item in normalized_labels), default=1
        )
        self._global_label_rank = {}

        training_examples = []
        for text, item_labels in zip(texts, normalized_labels):
            doc = self.nlp.make_doc(text)
            score_map = self._build_score_map(categories, item_labels)
            training_examples.append(Example.from_dict(doc, {"cats": score_map}))

        self.nlp.initialize()
        optimizer = self.nlp.create_optimizer()
        for _ in range(self.epochs):
            losses = {}
            self.nlp.update(training_examples, sgd=optimizer, losses=losses)

        return self

    def _rank_score_row(self, score_row: dict[str, float]) -> dict[str, float]:
        """Rank a single score row according to the configured scoring policy."""
        if not score_row:
            return {}

        if self.score_ranking == "document":
            return {
                label: score
                for label, score in sorted(
                    score_row.items(), key=lambda item: item[1], reverse=True
                )
            }

        global_rank = getattr(self, "_global_label_rank", {})
        return {
            label: score
            for label, score in sorted(
                score_row.items(),
                key=lambda item: (
                    item[1],
                    -global_rank.get(item[0], 0),
                ),
                reverse=True,
            )
        }

    def _build_global_label_rank(
        self, score_rows: Sequence[dict[str, float]]
    ) -> dict[str, int]:
        """Compute corpus-level label ordering from a collection of score rows."""
        if not score_rows:
            return {}

        label_scores: dict[str, list[float]] = {}
        for row in score_rows:
            for label, score in row.items():
                label_scores.setdefault(label, []).append(float(score))

        return {
            label: rank
            for rank, label in enumerate(
                sorted(
                    label_scores,
                    key=lambda label: (
                        sum(label_scores[label]) / len(label_scores[label]),
                        label,
                    ),
                    reverse=True,
                )
            )
        }

    def _score_rows_for_texts(self, texts: Sequence[str]) -> list[dict[str, float]]:
        """Return the score map for each document produced by the spaCy model."""
        return [
            {str(label): float(score) for label, score in (doc.cats or {}).items()}
            for doc in self.nlp.pipe(texts)
        ]

    def predict_scores(self, data: Any) -> list[dict[str, float]]:
        """Return the raw spaCy confidence scores attached to each predicted document.

        Args:
            data: The input data, which can be raw text, a standardized dataset, or Lexos docs.

        Returns:
            A list of dictionaries containing the raw confidence scores for each label.
        """
        texts = self._coerce_texts_from_input(data)
        score_rows = self._score_rows_for_texts(texts)
        if self.score_ranking == "global":
            self._global_label_rank = self._build_global_label_rank(score_rows)
        else:
            self._global_label_rank = {}

        return [self._rank_score_row(row) for row in score_rows]

    @staticmethod
    def _select_single_label(scores: dict[str, float]) -> str:
        """Return the strongest single-label prediction."""
        if not scores:
            return ""
        return max(scores, key=scores.get)

    def _select_multi_label_predictions(self, scores: dict[str, float]) -> list[str]:
        """Return the highest-scoring labels according to the configured ranking mode."""
        if not scores:
            return []

        target_count = max(1, getattr(self, "_target_label_count", 1))
        if self.score_ranking == "document":
            ranked_labels = [
                label
                for label, _ in sorted(
                    scores.items(), key=lambda item: item[1], reverse=True
                )
            ]
            return ranked_labels[:target_count]

        global_rank = getattr(self, "_global_label_rank", {})
        ranked_labels = [
            label
            for label in sorted(
                scores,
                key=lambda label: (scores[label], -global_rank.get(label, 0)),
                reverse=True,
            )
        ]
        return ranked_labels[:target_count]

    def _predict_document(self, doc: Any) -> list[str] | str:
        """Predict a label or a set of labels for one spaCy document result."""
        scores = doc.cats or {}
        if not scores:
            return [] if not self.exclusive_classes else ""
        if self.exclusive_classes:
            return self._select_single_label(scores)
        return self._select_multi_label_predictions(scores)

    def predict(self, data: Any) -> list[list[str] | str]:
        """Predict labels for raw text or tokenized Lexos documents.

        Args:
            data: The input data, which can be raw text, a standardized dataset, or Lexos docs.

        Returns:
            A list of predicted labels for each document.
        """
        texts = self._coerce_texts_from_input(data)
        doc_iter = list(self.nlp.pipe(texts))
        if self.score_ranking == "global":
            score_rows = [
                {str(label): float(score) for label, score in (doc.cats or {}).items()}
                for doc in doc_iter
            ]
            self._global_label_rank = self._build_global_label_rank(score_rows)
        else:
            self._global_label_rank = {}
        return [self._predict_document(doc) for doc in doc_iter]

    @staticmethod
    def _matches_predicted_labels(
        predicted: list[str] | str, gold_labels: set[str]
    ) -> bool:
        """Return whether a prediction matches the gold label set."""
        if isinstance(predicted, list):
            return set(predicted) == gold_labels
        if not predicted:
            return not gold_labels
        return predicted == next(iter(gold_labels), "")

    def evaluate(self, data: Any, labels: Sequence[Any]) -> dict[str, float]:
        """Evaluate the trained pipeline using accuracy.

        Args:
            data: The input text data to evaluate.
            labels: A sequence of true label strings or per-document label lists.

        Returns:
            A dictionary containing the accuracy of the predictions.
        """
        if not labels:
            return {"accuracy": 0.0}

        predictions = self.predict(data)
        matches = 0
        for predicted, label in zip(predictions, labels):
            gold_labels = set(self._coerce_label_list(label))
            if self._matches_predicted_labels(predicted, gold_labels):
                matches += 1
        return {"accuracy": matches / len(labels)}

architecture: str | dict[str, Any] = 'bow' pydantic-field

Text categorizer architecture or custom config.

epochs: int = 10 pydantic-field

Number of training epochs.

exclusive_classes: bool = True pydantic-field

Whether classes are mutually exclusive.

language: str = 'en' pydantic-field

SpaCy language shortcut to use.

model: Language property

Return the underlying spaCy language model.

Returns:

Type	Description
Language	The underlying spaCy Language object representing the text categorizer model.

name: str = 'classifier' pydantic-field

Human-readable pipeline name.

nlp: Language | None = None pydantic-field

Underlying spaCy language object.

score_ranking: Literal['document', 'global'] = 'document' pydantic-field

How to rank labels when selecting a multi-label prediction: 'document' ranks within each document, 'global' ranks using the global score ordering across all predicted documents.

__call__(data: Any) -> Sequence[str]

Convenience wrapper for predicting on a single data payload.

Source code in lexos/classification/classifier.py

def __call__(self, data: Any) -> Sequence[str]:
    """Convenience wrapper for predicting on a single data payload."""
    return self.predict(data)

__init__(**data)

Initialize the SpaCyTextCategorizerPipeline with the specified settings.

Source code in lexos/classification/spacy_pipeline.py

def __init__(self, **data):
    """Initialize the SpaCyTextCategorizerPipeline with the specified settings."""
    super().__init__(**data)
    if self.nlp is None:
        self.nlp = spacy.blank(self.language)

    pipe_name = self._resolved_pipe_name
    if pipe_name not in self.nlp.pipe_names:
        cfg = self._build_pipe_config()
        self.nlp.add_pipe(pipe_name, config=cfg)

evaluate(data: Any, labels: Sequence[Any]) -> dict[str, float]

Evaluate the trained pipeline using accuracy.

Parameters:

Name	Type	Description	Default
data	Any	The input text data to evaluate.	required
labels	Sequence[Any]	A sequence of true label strings or per-document label lists.	required

Returns:

Type	Description
dict[str, float]	A dictionary containing the accuracy of the predictions.

Source code in lexos/classification/spacy_pipeline.py

def evaluate(self, data: Any, labels: Sequence[Any]) -> dict[str, float]:
    """Evaluate the trained pipeline using accuracy.

    Args:
        data: The input text data to evaluate.
        labels: A sequence of true label strings or per-document label lists.

    Returns:
        A dictionary containing the accuracy of the predictions.
    """
    if not labels:
        return {"accuracy": 0.0}

    predictions = self.predict(data)
    matches = 0
    for predicted, label in zip(predictions, labels):
        gold_labels = set(self._coerce_label_list(label))
        if self._matches_predicted_labels(predicted, gold_labels):
            matches += 1
    return {"accuracy": matches / len(labels)}

fit(data: Any, labels: Sequence[Any]) -> SpaCyTextCategorizerPipeline

Fit a spaCy text categorizer on raw text or tokenized Lexos docs.

Parameters:

Name	Type	Description	Default
data	Any	The input data, which can be raw text, a standardized dataset, or Lexos docs.	required
labels	Sequence[Any]	The corresponding labels for each document in the input data.	required

Returns:

Type	Description
SpaCyTextCategorizerPipeline	The fitted SpaCyTextCategorizerPipeline instance.

Source code in lexos/classification/spacy_pipeline.py

def fit(self, data: Any, labels: Sequence[Any]) -> "SpaCyTextCategorizerPipeline":
    """Fit a spaCy text categorizer on raw text or tokenized Lexos docs.

    Args:
        data: The input data, which can be raw text, a standardized dataset, or Lexos docs.
        labels: The corresponding labels for each document in the input data.

    Returns:
        The fitted SpaCyTextCategorizerPipeline instance.
    """
    texts = self._coerce_texts_from_input(data)
    label_values = list(labels)

    if len(texts) != len(label_values):
        raise ValueError("Number of texts must match the number of labels.")

    normalized_labels = [self._coerce_label_list(label) for label in label_values]
    if self.exclusive_classes and any(len(item) > 1 for item in normalized_labels):
        self._switch_to_multi_label_mode()

    if self.exclusive_classes:
        for i, item in enumerate(normalized_labels):
            if len(item) != 1:
                raise ValueError(
                    "exclusive_classes=True requires exactly one label per document; "
                    f"found {item!r} for row {i}."
                )

    categories = self._ensure_labels(normalized_labels)
    if not categories:
        raise ValueError("At least one label is required before training.")

    self._target_label_count = max(
        (len(item) for item in normalized_labels), default=1
    )
    self._global_label_rank = {}

    training_examples = []
    for text, item_labels in zip(texts, normalized_labels):
        doc = self.nlp.make_doc(text)
        score_map = self._build_score_map(categories, item_labels)
        training_examples.append(Example.from_dict(doc, {"cats": score_map}))

    self.nlp.initialize()
    optimizer = self.nlp.create_optimizer()
    for _ in range(self.epochs):
        losses = {}
        self.nlp.update(training_examples, sgd=optimizer, losses=losses)

    return self

load(path: str | Path) -> SpaCyTextCategorizerPipeline classmethod

Load a saved spaCy pipeline instance and restore its configuration.

Source code in lexos/classification/spacy_pipeline.py

@classmethod
def load(cls, path: str | Path) -> "SpaCyTextCategorizerPipeline":
    """Load a saved spaCy pipeline instance and restore its configuration."""
    target = Path(path)
    config = json.loads((target / "pipeline_config.json").read_text())
    pipeline = cls(**{k: v for k, v in config.items() if not k.startswith("_")})
    pipeline.nlp = spacy.load(str(target / "model"))
    pipeline._labels = list(config.get("_labels", []))
    pipeline._target_label_count = int(config.get("_target_label_count", 1))
    pipeline._global_label_rank = dict(config.get("_global_label_rank", {}))
    pipeline._resolved_pipe_name = (
        "textcat_multilabel" if not pipeline.exclusive_classes else "textcat"
    )
    return pipeline

predict(data: Any) -> list[list[str] | str]

Predict labels for raw text or tokenized Lexos documents.

Parameters:

Name	Type	Description	Default
data	Any	The input data, which can be raw text, a standardized dataset, or Lexos docs.	required

Returns:

Type	Description
list[list[str] | str]	A list of predicted labels for each document.

Source code in lexos/classification/spacy_pipeline.py

def predict(self, data: Any) -> list[list[str] | str]:
    """Predict labels for raw text or tokenized Lexos documents.

    Args:
        data: The input data, which can be raw text, a standardized dataset, or Lexos docs.

    Returns:
        A list of predicted labels for each document.
    """
    texts = self._coerce_texts_from_input(data)
    doc_iter = list(self.nlp.pipe(texts))
    if self.score_ranking == "global":
        score_rows = [
            {str(label): float(score) for label, score in (doc.cats or {}).items()}
            for doc in doc_iter
        ]
        self._global_label_rank = self._build_global_label_rank(score_rows)
    else:
        self._global_label_rank = {}
    return [self._predict_document(doc) for doc in doc_iter]

predict_scores(data: Any) -> list[dict[str, float]]

Return the raw spaCy confidence scores attached to each predicted document.

Parameters:

Name	Type	Description	Default
data	Any	The input data, which can be raw text, a standardized dataset, or Lexos docs.	required

Returns:

Type	Description
list[dict[str, float]]	A list of dictionaries containing the raw confidence scores for each label.

Source code in lexos/classification/spacy_pipeline.py

def predict_scores(self, data: Any) -> list[dict[str, float]]:
    """Return the raw spaCy confidence scores attached to each predicted document.

    Args:
        data: The input data, which can be raw text, a standardized dataset, or Lexos docs.

    Returns:
        A list of dictionaries containing the raw confidence scores for each label.
    """
    texts = self._coerce_texts_from_input(data)
    score_rows = self._score_rows_for_texts(texts)
    if self.score_ranking == "global":
        self._global_label_rank = self._build_global_label_rank(score_rows)
    else:
        self._global_label_rank = {}

    return [self._rank_score_row(row) for row in score_rows]

save(path: str | Path) -> None

Save the trained spaCy pipeline and its configuration to disk.

Source code in lexos/classification/spacy_pipeline.py

def save(self, path: str | Path) -> None:
    """Save the trained spaCy pipeline and its configuration to disk."""
    target = Path(path)
    target.mkdir(parents=True, exist_ok=True)
    config = self.model_dump(mode="python", exclude={"nlp"})
    config["_labels"] = list(self._labels)
    config["_target_label_count"] = int(getattr(self, "_target_label_count", 1))
    config["_global_label_rank"] = dict(getattr(self, "_global_label_rank", {}))
    config_path = target / "pipeline_config.json"
    config_path.write_text(json.dumps(config, indent=2))
    self.nlp.to_disk(str(target / "model"))

rendering:
  show_root_heading: true
  heading_level: 3

Classifier pydantic-model

Bases: BaseModel

High-level classification orchestration for non-technical users.

The Classifier class is intentionally backend-agnostic. Users supply data, labels, and a pipeline object that implements the backend-specific training and inference logic. This keeps the public API stable across SpaCy, scikit-learn, and other classification methods.

Config:

• arbitrary_types_allowed: True

Fields:

• data (Any | None)
• labels (Sequence[Any])
• titles (Sequence[Any])
• pipeline (BaseClassificationPipeline | None)

Validators:

• _validate_pipeline

Source code in lexos/classification/classifier.py

class Classifier(BaseModel):
    """High-level classification orchestration for non-technical users.

    The `Classifier` class is intentionally backend-agnostic. Users supply data,
    labels, and a pipeline object that implements the backend-specific training and
    inference logic. This keeps the public API stable across SpaCy, scikit-learn,
    and other classification methods.
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)

    data: Any | None = Field(default=None, description="Training or prediction data.")
    labels: Sequence[Any] = Field(
        default_factory=list, description="Classification labels."
    )
    titles: Sequence[Any] = Field(
        default_factory=list, description="Document titles aligned with the data rows."
    )
    pipeline: BaseClassificationPipeline | None = Field(
        default=None,
        description="Classification backend strategy; e.g. a spaCy or scikit-learn pipeline.",
    )

    _fitted: bool = PrivateAttr(default=False)
    _last_fit_data: Any = PrivateAttr(default=None)
    _last_fit_labels: list[Any] = PrivateAttr(default_factory=list)

    @model_validator(mode="after")
    def _validate_pipeline(self) -> "Classifier":
        """Validate that the pipeline is correctly set and an instance of BaseClassificationPipeline.

        Returns:
            The validated Classifier instance.

        Raises:
            TypeError: If the pipeline is not an instance of BaseClassificationPipeline.
        """
        if self.pipeline is not None and not isinstance(
            self.pipeline, BaseClassificationPipeline
        ):
            raise TypeError(
                "pipeline must be an instance of BaseClassificationPipeline."
            )
        return self

    def _resolve_data_and_labels(
        self,
        data: Any | None = None,
        labels: Sequence[Any] | None = None,
        titles: Sequence[Any] | None = None,
    ) -> tuple[ClassifierData, list[Any]]:
        """Resolve the input data and labels, normalizing them into a ClassifierData instance.

        Args:
            data: The input data to resolve. If None, the stored data is used.
            labels: The input labels to resolve. If None, the stored labels are used.
            titles: Optional titles aligned with the input rows.

        Returns:
            A tuple containing the normalized ClassifierData instance and the corresponding list of labels.
        """
        resolved_data = self.data if data is None else data
        explicit_labels = list(labels) if labels is not None else None
        if explicit_labels is None:
            explicit_labels = list(self.labels) if self.labels else None

        explicit_titles = list(titles) if titles is not None else None
        normalized = ClassifierData.from_input(
            resolved_data,
            explicit_labels,
            titles=explicit_titles,
        )

        if explicit_titles is None and self.titles:
            if len(self.titles) == normalized.row_count():
                normalized.titles = list(self.titles)

        return normalized, normalized.labels

    def fit(
        self, data: Any | None = None, labels: Sequence[Any] | None = None
    ) -> "Classifier":
        """Fit the classifier using the supplied pipeline backend.

        Args:
            data: The input data to fit the classifier on. If None, the stored data is used.
            labels: The corresponding labels for the input data. If None, the stored labels are used.

        Returns:
            The fitted Classifier instance.
        """
        if self.pipeline is None:
            raise ValueError(
                "A classification pipeline must be configured before calling fit()."
            )

        resolved_data, resolved_labels = self._resolve_data_and_labels(data, labels)
        self.pipeline.fit(resolved_data, resolved_labels)
        self.data = resolved_data.values
        self.labels = resolved_labels
        self.titles = (
            list(resolved_data.titles) if resolved_data.titles is not None else []
        )
        self._fitted = True
        self._last_fit_data = resolved_data
        self._last_fit_labels = resolved_labels
        return self

    def predict(self, data: Any | None = None) -> list[str]:
        """Predict labels for the supplied data or for the stored training data.

        Args:
            data: The input data to predict labels for. If None, the stored data is used.

        Returns:
            A list of predicted labels for each document.
        """
        if self.pipeline is None:
            raise ValueError(
                "A classification pipeline must be configured before calling predict()."
            )

        if data is None:
            if self.data is None:
                raise ValueError(
                    "No prediction data was supplied and no fitted data is available."
                )
            data = self.data

        predictions = self.pipeline.predict(data)
        return [
            value if isinstance(value, (list, tuple, set)) else str(value)
            for value in predictions
        ]

    def predict_scores(self, data: Any | None = None) -> list[dict[str, float]]:
        """Return the underlying confidence scores for each prediction when available.

        Args:
            data: The input data to predict scores for. If None, the stored data is used.

        Returns:
            A list of dictionaries containing confidence scores for each prediction.
        """
        if self.pipeline is None:
            raise ValueError(
                "A classification pipeline must be configured before calling predict_scores()."
            )

        if data is None:
            if self.data is None:
                raise ValueError(
                    "No prediction data was supplied and no fitted data is available."
                )
            data = self.data

        if not hasattr(self.pipeline, "predict_scores"):
            raise NotImplementedError(
                f"{type(self.pipeline).__name__} does not implement predict_scores()."
            )

        scores = self.pipeline.predict_scores(data)
        # Sort the scores
        for row in scores:
            row.update(
                dict(sorted(row.items(), key=lambda item: item[1], reverse=True))
            )
        return scores

    def evaluate(
        self, data: Any | None = None, labels: Sequence[Any] | None = None
    ) -> dict[str, float]:
        """Evaluate the fitted pipeline on the supplied data and labels.

        Args:
            data: The input data to evaluate the pipeline on. If None, the stored data is used.
            labels: The corresponding labels for the input data. If None, the stored labels are used.

        Returns:
            A dictionary containing evaluation metrics for the predictions.
        """
        if self.pipeline is None:
            raise ValueError(
                "A classification pipeline must be configured before calling evaluate()."
            )

        if data is None:
            if self.data is None:
                raise ValueError(
                    "No evaluation data was supplied and no fitted data is available."
                )
            data = self.data
        if labels is None:
            if not self.labels:
                raise ValueError("No labels were provided for evaluation.")
            labels = self.labels

        return self.pipeline.evaluate(data, list(labels))

    def split_data(
        self,
        data: Any | None = None,
        labels: Sequence[str] | None = None,
        titles: Sequence[Any] | None = None,
        test_size: float = 0.2,
        dev_size: float | None = None,
        random_state: int = 42,
        stratify: bool = True,
    ) -> dict[str, Any]:
        """Split data into train/test/dev partitions.

        Args:
            data: data to split; defaults to the classifier's stored data.
            labels: labels aligned to the data; defaults to the classifier's labels.
            titles: optional titles aligned to the rows; preserved in the output.
            test_size: fraction of the data reserved for testing.
            dev_size: optional fraction reserved for development / validation.
            random_state: deterministic random seed.
            stratify: whether to preserve label distributions across splits.

        Returns:
            Dictionary containing the partitions keyed by `train`, `test`, and optional
            `dev` data plus the corresponding labels and titles.
        """
        resolved_data, resolved_labels = self._resolve_data_and_labels(
            data,
            labels,
            titles=titles,
        )

        split = resolved_data.split(
            test_size=test_size,
            dev_size=dev_size,
            random_state=random_state,
            stratify=stratify,
        )

        result = {
            "train": {
                "data": split["train"].values,
                "labels": split["train"].labels,
                "titles": split["train"].titles,
            },
            "test": {
                "data": split["test"].values,
                "labels": split["test"].labels,
                "titles": split["test"].titles,
            },
        }
        if "dev" in split:
            result["dev"] = {
                "data": split["dev"].values,
                "labels": split["dev"].labels,
                "titles": split["dev"].titles,
            }
        return result

    def train_test_split(
        self,
        test_size: float = 0.2,
        random_state: int = 42,
        stratify: bool = True,
        titles: Sequence[Any] | None = None,
    ) -> dict[str, Any]:
        """Convenience wrapper for train/test splitting.

        Args:
            test_size: fraction of the data reserved for testing.
            random_state: deterministic random seed.
            stratify: whether to preserve label distributions across splits.
            titles: optional titles aligned with the rows; preserved in the output.

        Returns:
            Dictionary containing the train and test partitions keyed by `train` and `test` data plus the corresponding labels and titles.
        """
        return self.split_data(
            titles=titles,
            test_size=test_size,
            random_state=random_state,
            stratify=stratify,
        )

    def train_dev_split(
        self,
        dev_size: float = 0.2,
        random_state: int = 42,
        stratify: bool = True,
        titles: Sequence[Any] | None = None,
    ) -> dict[str, Any]:
        """Convenience wrapper for train/dev splitting.

        Args:
            dev_size: fraction of the data reserved for development / validation.
            random_state: deterministic random seed.
            stratify: whether to preserve label distributions across splits.
            titles: optional titles aligned with the rows; preserved in the output.

        Returns:
            Dictionary containing the train and dev partitions keyed by `train` and `dev` data plus the corresponding labels and titles.
        """
        return self.split_data(
            titles=titles,
            dev_size=dev_size,
            random_state=random_state,
            stratify=stratify,
        )

    def __call__(self, data: Any) -> list[str]:
        """Convenience wrapper for prediction calls.

        Args:
            data: The input data to make predictions on.

        Returns:
            A list of predicted labels for the input data.
        """
        return self.predict(data)

data: Any | None = None pydantic-field

Training or prediction data.

labels: Sequence[Any] pydantic-field

Classification labels.

pipeline: BaseClassificationPipeline | None = None pydantic-field

Classification backend strategy; e.g. a spaCy or scikit-learn pipeline.

titles: Sequence[Any] pydantic-field

Document titles aligned with the data rows.

__call__(data: Any) -> list[str]

Convenience wrapper for prediction calls.

Parameters:

Name	Type	Description	Default
data	Any	The input data to make predictions on.	required

Returns:

Type	Description
list[str]	A list of predicted labels for the input data.

Source code in lexos/classification/classifier.py

def __call__(self, data: Any) -> list[str]:
    """Convenience wrapper for prediction calls.

    Args:
        data: The input data to make predictions on.

    Returns:
        A list of predicted labels for the input data.
    """
    return self.predict(data)

evaluate(data: Any | None = None, labels: Sequence[Any] | None = None) -> dict[str, float]

Evaluate the fitted pipeline on the supplied data and labels.

Parameters:

Name	Type	Description	Default
data	Any | None	The input data to evaluate the pipeline on. If None, the stored data is used.	None
labels	Sequence[Any] | None	The corresponding labels for the input data. If None, the stored labels are used.	None

Returns:

Type	Description
dict[str, float]	A dictionary containing evaluation metrics for the predictions.

Source code in lexos/classification/classifier.py

def evaluate(
    self, data: Any | None = None, labels: Sequence[Any] | None = None
) -> dict[str, float]:
    """Evaluate the fitted pipeline on the supplied data and labels.

    Args:
        data: The input data to evaluate the pipeline on. If None, the stored data is used.
        labels: The corresponding labels for the input data. If None, the stored labels are used.

    Returns:
        A dictionary containing evaluation metrics for the predictions.
    """
    if self.pipeline is None:
        raise ValueError(
            "A classification pipeline must be configured before calling evaluate()."
        )

    if data is None:
        if self.data is None:
            raise ValueError(
                "No evaluation data was supplied and no fitted data is available."
            )
        data = self.data
    if labels is None:
        if not self.labels:
            raise ValueError("No labels were provided for evaluation.")
        labels = self.labels

    return self.pipeline.evaluate(data, list(labels))

fit(data: Any | None = None, labels: Sequence[Any] | None = None) -> Classifier

Fit the classifier using the supplied pipeline backend.

Parameters:

Name	Type	Description	Default
data	Any | None	The input data to fit the classifier on. If None, the stored data is used.	None
labels	Sequence[Any] | None	The corresponding labels for the input data. If None, the stored labels are used.	None

Returns:

Type	Description
Classifier	The fitted Classifier instance.

Source code in lexos/classification/classifier.py

def fit(
    self, data: Any | None = None, labels: Sequence[Any] | None = None
) -> "Classifier":
    """Fit the classifier using the supplied pipeline backend.

    Args:
        data: The input data to fit the classifier on. If None, the stored data is used.
        labels: The corresponding labels for the input data. If None, the stored labels are used.

    Returns:
        The fitted Classifier instance.
    """
    if self.pipeline is None:
        raise ValueError(
            "A classification pipeline must be configured before calling fit()."
        )

    resolved_data, resolved_labels = self._resolve_data_and_labels(data, labels)
    self.pipeline.fit(resolved_data, resolved_labels)
    self.data = resolved_data.values
    self.labels = resolved_labels
    self.titles = (
        list(resolved_data.titles) if resolved_data.titles is not None else []
    )
    self._fitted = True
    self._last_fit_data = resolved_data
    self._last_fit_labels = resolved_labels
    return self

predict(data: Any | None = None) -> list[str]

Predict labels for the supplied data or for the stored training data.

Parameters:

Name	Type	Description	Default
data	Any | None	The input data to predict labels for. If None, the stored data is used.	None

Returns:

Type	Description
list[str]	A list of predicted labels for each document.

Source code in lexos/classification/classifier.py

def predict(self, data: Any | None = None) -> list[str]:
    """Predict labels for the supplied data or for the stored training data.

    Args:
        data: The input data to predict labels for. If None, the stored data is used.

    Returns:
        A list of predicted labels for each document.
    """
    if self.pipeline is None:
        raise ValueError(
            "A classification pipeline must be configured before calling predict()."
        )

    if data is None:
        if self.data is None:
            raise ValueError(
                "No prediction data was supplied and no fitted data is available."
            )
        data = self.data

    predictions = self.pipeline.predict(data)
    return [
        value if isinstance(value, (list, tuple, set)) else str(value)
        for value in predictions
    ]

predict_scores(data: Any | None = None) -> list[dict[str, float]]

Return the underlying confidence scores for each prediction when available.

Parameters:

Name	Type	Description	Default
data	Any | None	The input data to predict scores for. If None, the stored data is used.	None

Returns:

Type	Description
list[dict[str, float]]	A list of dictionaries containing confidence scores for each prediction.

Source code in lexos/classification/classifier.py

def predict_scores(self, data: Any | None = None) -> list[dict[str, float]]:
    """Return the underlying confidence scores for each prediction when available.

    Args:
        data: The input data to predict scores for. If None, the stored data is used.

    Returns:
        A list of dictionaries containing confidence scores for each prediction.
    """
    if self.pipeline is None:
        raise ValueError(
            "A classification pipeline must be configured before calling predict_scores()."
        )

    if data is None:
        if self.data is None:
            raise ValueError(
                "No prediction data was supplied and no fitted data is available."
            )
        data = self.data

    if not hasattr(self.pipeline, "predict_scores"):
        raise NotImplementedError(
            f"{type(self.pipeline).__name__} does not implement predict_scores()."
        )

    scores = self.pipeline.predict_scores(data)
    # Sort the scores
    for row in scores:
        row.update(
            dict(sorted(row.items(), key=lambda item: item[1], reverse=True))
        )
    return scores

split_data(data: Any | None = None, labels: Sequence[str] | None = None, titles: Sequence[Any] | None = None, test_size: float = 0.2, dev_size: float | None = None, random_state: int = 42, stratify: bool = True) -> dict[str, Any]

Split data into train/test/dev partitions.

Parameters:

Name	Type	Description	Default
data	Any | None	data to split; defaults to the classifier's stored data.	None
labels	Sequence[str] | None	labels aligned to the data; defaults to the classifier's labels.	None
titles	Sequence[Any] | None	optional titles aligned to the rows; preserved in the output.	None
test_size	float	fraction of the data reserved for testing.	0.2
dev_size	float | None	optional fraction reserved for development / validation.	None
random_state	int	deterministic random seed.	42
stratify	bool	whether to preserve label distributions across splits.	True

Returns:

Type	Description
dict[str, Any]	Dictionary containing the partitions keyed by train, test, and optional
dict[str, Any]	dev data plus the corresponding labels and titles.

Source code in lexos/classification/classifier.py

def split_data(
    self,
    data: Any | None = None,
    labels: Sequence[str] | None = None,
    titles: Sequence[Any] | None = None,
    test_size: float = 0.2,
    dev_size: float | None = None,
    random_state: int = 42,
    stratify: bool = True,
) -> dict[str, Any]:
    """Split data into train/test/dev partitions.

    Args:
        data: data to split; defaults to the classifier's stored data.
        labels: labels aligned to the data; defaults to the classifier's labels.
        titles: optional titles aligned to the rows; preserved in the output.
        test_size: fraction of the data reserved for testing.
        dev_size: optional fraction reserved for development / validation.
        random_state: deterministic random seed.
        stratify: whether to preserve label distributions across splits.

    Returns:
        Dictionary containing the partitions keyed by `train`, `test`, and optional
        `dev` data plus the corresponding labels and titles.
    """
    resolved_data, resolved_labels = self._resolve_data_and_labels(
        data,
        labels,
        titles=titles,
    )

    split = resolved_data.split(
        test_size=test_size,
        dev_size=dev_size,
        random_state=random_state,
        stratify=stratify,
    )

    result = {
        "train": {
            "data": split["train"].values,
            "labels": split["train"].labels,
            "titles": split["train"].titles,
        },
        "test": {
            "data": split["test"].values,
            "labels": split["test"].labels,
            "titles": split["test"].titles,
        },
    }
    if "dev" in split:
        result["dev"] = {
            "data": split["dev"].values,
            "labels": split["dev"].labels,
            "titles": split["dev"].titles,
        }
    return result

train_dev_split(dev_size: float = 0.2, random_state: int = 42, stratify: bool = True, titles: Sequence[Any] | None = None) -> dict[str, Any]

Convenience wrapper for train/dev splitting.

Parameters:

Name	Type	Description	Default
dev_size	float	fraction of the data reserved for development / validation.	0.2
random_state	int	deterministic random seed.	42
stratify	bool	whether to preserve label distributions across splits.	True
titles	Sequence[Any] | None	optional titles aligned with the rows; preserved in the output.	None

Returns:

Type	Description
dict[str, Any]	Dictionary containing the train and dev partitions keyed by train and dev data plus the corresponding labels and titles.

Source code in lexos/classification/classifier.py

def train_dev_split(
    self,
    dev_size: float = 0.2,
    random_state: int = 42,
    stratify: bool = True,
    titles: Sequence[Any] | None = None,
) -> dict[str, Any]:
    """Convenience wrapper for train/dev splitting.

    Args:
        dev_size: fraction of the data reserved for development / validation.
        random_state: deterministic random seed.
        stratify: whether to preserve label distributions across splits.
        titles: optional titles aligned with the rows; preserved in the output.

    Returns:
        Dictionary containing the train and dev partitions keyed by `train` and `dev` data plus the corresponding labels and titles.
    """
    return self.split_data(
        titles=titles,
        dev_size=dev_size,
        random_state=random_state,
        stratify=stratify,
    )

train_test_split(test_size: float = 0.2, random_state: int = 42, stratify: bool = True, titles: Sequence[Any] | None = None) -> dict[str, Any]

Convenience wrapper for train/test splitting.

Parameters:

Name	Type	Description	Default
test_size	float	fraction of the data reserved for testing.	0.2
random_state	int	deterministic random seed.	42
stratify	bool	whether to preserve label distributions across splits.	True
titles	Sequence[Any] | None	optional titles aligned with the rows; preserved in the output.	None

Returns:

Type	Description
dict[str, Any]	Dictionary containing the train and test partitions keyed by train and test data plus the corresponding labels and titles.

Source code in lexos/classification/classifier.py

def train_test_split(
    self,
    test_size: float = 0.2,
    random_state: int = 42,
    stratify: bool = True,
    titles: Sequence[Any] | None = None,
) -> dict[str, Any]:
    """Convenience wrapper for train/test splitting.

    Args:
        test_size: fraction of the data reserved for testing.
        random_state: deterministic random seed.
        stratify: whether to preserve label distributions across splits.
        titles: optional titles aligned with the rows; preserved in the output.

    Returns:
        Dictionary containing the train and test partitions keyed by `train` and `test` data plus the corresponding labels and titles.
    """
    return self.split_data(
        titles=titles,
        test_size=test_size,
        random_state=random_state,
        stratify=stratify,
    )

rendering:
  show_root_heading: true
  heading_level: 3

BaseClassificationPipeline pydantic-model

Bases: BaseModel

Abstract strategy interface for classification backends.

Subclasses implement the concrete logic for each method, such as spaCy TextCategorizer or a scikit-learn estimator.

Config:

• arbitrary_types_allowed: True

Fields:

• name (str)

Source code in lexos/classification/classifier.py

class BaseClassificationPipeline(BaseModel):
    """Abstract strategy interface for classification backends.

    Subclasses implement the concrete logic for each method, such as spaCy
    `TextCategorizer` or a scikit-learn estimator.
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)
    name: str = Field(default="classifier", description="Human-readable pipeline name.")

    @property
    def model(self) -> Any:
        """Return the underlying backend model object.

        Returns:
            The underlying backend model object.
        """
        raise NotImplementedError

    def fit(self, data: Any, labels: Sequence[str]) -> Any:
        """Train the pipeline on the supplied data and labels.

        Args:
            data: The input data to train on.
            labels: The corresponding labels for the input data.

        Returns:
            The trained pipeline instance.
        """
        raise NotImplementedError

    def predict(self, data: Any) -> Sequence[str]:
        """Predict labels for the supplied data.

        Args:
            data: The input data to make predictions on.

        Returns:
            A list of predicted labels for the input data.
        """
        raise NotImplementedError

    def predict_scores(self, data: Any) -> Sequence[dict[str, float]]:
        """Return the prediction probabilities or confidences for each input item.

        Args:
            data: The input data to make predictions on.

        Returns:
            A list of dictionaries containing prediction probabilities or confidences for each input item.
        """
        raise NotImplementedError

    def evaluate(self, data: Any, labels: Sequence[str]) -> dict[str, float]:
        """Evaluate the fitted pipeline on a dataset.

        Args:
            data: The input data to evaluate on.
            labels: The corresponding labels for the input data.

        Returns:
            A dictionary containing evaluation metrics for the input data and labels.
        """
        raise NotImplementedError

    def save(self, path: str | Any) -> None:
        """Persist the fitted pipeline and its configuration to disk."""
        raise NotImplementedError

    @classmethod
    def load(cls, path: str | Any) -> "BaseClassificationPipeline":
        """Load a saved pipeline instance from disk."""
        raise NotImplementedError

    def __call__(self, data: Any) -> Sequence[str]:
        """Convenience wrapper for predicting on a single data payload."""
        return self.predict(data)

model: Any property

Return the underlying backend model object.

Returns:

Type	Description
Any	The underlying backend model object.

name: str = 'classifier' pydantic-field

Human-readable pipeline name.

__call__(data: Any) -> Sequence[str]

Convenience wrapper for predicting on a single data payload.

Source code in lexos/classification/classifier.py

def __call__(self, data: Any) -> Sequence[str]:
    """Convenience wrapper for predicting on a single data payload."""
    return self.predict(data)

evaluate(data: Any, labels: Sequence[str]) -> dict[str, float]

Evaluate the fitted pipeline on a dataset.

Parameters:

Name	Type	Description	Default
data	Any	The input data to evaluate on.	required
labels	Sequence[str]	The corresponding labels for the input data.	required

Returns:

Type	Description
dict[str, float]	A dictionary containing evaluation metrics for the input data and labels.

Source code in lexos/classification/classifier.py

def evaluate(self, data: Any, labels: Sequence[str]) -> dict[str, float]:
    """Evaluate the fitted pipeline on a dataset.

    Args:
        data: The input data to evaluate on.
        labels: The corresponding labels for the input data.

    Returns:
        A dictionary containing evaluation metrics for the input data and labels.
    """
    raise NotImplementedError

fit(data: Any, labels: Sequence[str]) -> Any

Train the pipeline on the supplied data and labels.

Parameters:

Name	Type	Description	Default
data	Any	The input data to train on.	required
labels	Sequence[str]	The corresponding labels for the input data.	required

Returns:

Type	Description
Any	The trained pipeline instance.

Source code in lexos/classification/classifier.py

def fit(self, data: Any, labels: Sequence[str]) -> Any:
    """Train the pipeline on the supplied data and labels.

    Args:
        data: The input data to train on.
        labels: The corresponding labels for the input data.

    Returns:
        The trained pipeline instance.
    """
    raise NotImplementedError

load(path: str | Any) -> BaseClassificationPipeline classmethod

Load a saved pipeline instance from disk.

Source code in lexos/classification/classifier.py

@classmethod
def load(cls, path: str | Any) -> "BaseClassificationPipeline":
    """Load a saved pipeline instance from disk."""
    raise NotImplementedError

predict(data: Any) -> Sequence[str]

Predict labels for the supplied data.

Parameters:

Name	Type	Description	Default
data	Any	The input data to make predictions on.	required

Returns:

Type	Description
Sequence[str]	A list of predicted labels for the input data.

Source code in lexos/classification/classifier.py

def predict(self, data: Any) -> Sequence[str]:
    """Predict labels for the supplied data.

    Args:
        data: The input data to make predictions on.

    Returns:
        A list of predicted labels for the input data.
    """
    raise NotImplementedError

predict_scores(data: Any) -> Sequence[dict[str, float]]

Return the prediction probabilities or confidences for each input item.

Parameters:

Name	Type	Description	Default
data	Any	The input data to make predictions on.	required

Returns:

Type	Description
Sequence[dict[str, float]]	A list of dictionaries containing prediction probabilities or confidences for each input item.

Source code in lexos/classification/classifier.py

def predict_scores(self, data: Any) -> Sequence[dict[str, float]]:
    """Return the prediction probabilities or confidences for each input item.

    Args:
        data: The input data to make predictions on.

    Returns:
        A list of dictionaries containing prediction probabilities or confidences for each input item.
    """
    raise NotImplementedError

save(path: str | Any) -> None

Persist the fitted pipeline and its configuration to disk.

Source code in lexos/classification/classifier.py

def save(self, path: str | Any) -> None:
    """Persist the fitted pipeline and its configuration to disk."""
    raise NotImplementedError

rendering:
  show_root_heading: true
  heading_level: 3

ClassifierData pydantic-model

Bases: BaseModel

Standardized input wrapper for training and prediction data.

This object centralizes the data-shape concerns ensuring that data is handled consistently before it is passed to Classifier.

Config:

• arbitrary_types_allowed: True

Fields:

• values (Any)
• labels (list[Any])
• docs (Any)
• titles (list[Any] | None)
• matrix (Any)
• source (str)

Source code in lexos/classification/classifier.py

class ClassifierData(BaseModel):
    """Standardized input wrapper for training and prediction data.

    This object centralizes the data-shape concerns ensuring that data is handled consistently before it is passed to `Classifier`.
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)

    values: Any
    labels: list[Any] = Field(default_factory=list)
    docs: Any = None
    titles: list[Any] | None = None
    matrix: Any = None
    source: str = "raw"

    def __init__(
        self,
        values: Any,
        labels: Sequence[Any] | None = None,
        *,
        docs: Any = None,
        titles: Sequence[Any] | None = None,
        matrix: Any = None,
        source: str = "raw",
    ) -> None:
        """Initialize the ClassifierData object.

        Args:
            values: The main data values.
            labels: Optional sequence of labels corresponding to the data.
            docs: Optional sequence of document objects.
            titles: Optional sequence of titles for the data items.
            matrix: Optional matrix representation of the data.
            source: A string indicating the source of the data.
        """
        super().__init__(
            values=values,
            labels=list(labels) if labels is not None else [],
            docs=docs,
            titles=list(titles) if titles is not None else None,
            matrix=matrix,
            source=source,
        )

    @staticmethod
    def _matrix_row_count(matrix: Any) -> int:
        """Return the number of rows in a matrix-like object, including list-backed inputs.

        Args:
            matrix: The matrix-like object to count rows for.

        Returns:
            The number of rows in the matrix-like object.
        """
        shape = getattr(matrix, "shape", None)
        if shape is not None:
            return int(shape[0])
        return len(matrix)

    @staticmethod
    def _validate_label_count(
        expected_count: int | None, actual_count: int, context: str
    ) -> None:
        """Raise a ValueError when label length diverges from the data shape.

        Args:
            expected_count: The expected number of labels.
            actual_count: The actual number of labels.
            context: A message to include in the ValueError if the counts do not match.

        Raises:
            ValueError: If the expected count is not None and does not match the actual count.
        """
        if expected_count is not None and expected_count != actual_count:
            raise ValueError(f"{context}")

    @classmethod
    def _from_dtm_input(
        cls,
        data: Any,
        labels: Sequence[Any] | None = None,
        titles: Sequence[Any] | None = None,
    ) -> "ClassifierData":
        """Normalize a Lexos DTM object into a standardized Dataset wrapper.

        Args:
            data: The Lexos DTM object to normalize.
            labels: Optional sequence of labels to associate with the data. If not provided, labels will be inferred from the DTM object.
            titles: Optional sequence of titles aligned with the records.

        Returns:
            An instance of the ClassifierData class wrapping the normalized data.
        """
        matrix = getattr(data, "doc_term_matrix", None)
        docs = getattr(data, "docs", None)
        resolved_labels = list(getattr(data, "labels", []) or [])
        resolved_titles = (
            list(getattr(data, "titles", []) or []) if hasattr(data, "titles") else []
        )
        if labels is not None and len(labels) > 0:
            resolved_labels = list(labels)
        if titles is not None and len(titles) > 0:
            resolved_titles = list(titles)

        if matrix is not None:
            row_count = cls._matrix_row_count(matrix)
        elif docs is not None:
            row_count = len(docs)
        else:
            row_count = len(resolved_labels)

        resolved_labels, resolved_titles = cls._resolve_row_alignment(
            row_count,
            resolved_labels,
            resolved_titles,
            "Label count must match the row count in the DTM document-term matrix."
            if matrix is not None
            else "Label count must match the number of stored DTM docs.",
            "Title count must match the number of rows in the dataset.",
        )

        return cls(
            values=matrix if matrix is not None else docs,
            labels=resolved_labels,
            docs=docs,
            titles=resolved_titles,
            matrix=matrix,
            source="dtm",
        )

    @classmethod
    def _from_dataframe_input(
        cls,
        data: pd.DataFrame,
        labels: Sequence[Any] | None = None,
        titles: Sequence[Any] | None = None,
    ) -> "ClassifierData":
        """Normalize a pandas DataFrame input into a standardized Dataset wrapper.

        Args:
            data: The pandas DataFrame to normalize.
            labels: Optional sequence of labels to associate with the data. If not provided, labels will be inferred from the DataFrame.

        Returns:
            An instance of the ClassifierData class wrapping the normalized data.
        """
        frame_labels = (
            list(data["label"].astype(str)) if "label" in data.columns else []
        )
        resolved_labels = list(frame_labels if labels is None else labels)
        resolved_titles = list(data["title"]) if "title" in data.columns else []
        if titles is not None and len(titles) > 0:
            resolved_titles = list(titles)

        resolved_labels, resolved_titles = cls._resolve_row_alignment(
            len(data),
            resolved_labels,
            resolved_titles,
            "Label count must match the row count in the DataFrame.",
            "Title count must match the row count in the DataFrame.",
        )

        return cls(
            values=data,
            labels=resolved_labels,
            titles=resolved_titles,
            source="dataframe",
        )

    @classmethod
    def _from_matrix_input(
        cls,
        data: Any,
        labels: Sequence[Any] | None = None,
        titles: Sequence[Any] | None = None,
    ) -> "ClassifierData":
        """Normalize matrix-like inputs into a standardized Dataset wrapper.

        Args:
            data: The matrix-like data to normalize.
            labels: Optional sequence of labels to associate with the data. If not provided, labels will be inferred from the matrix-like data.

        Returns:
            An instance of the ClassifierData class wrapping the normalized data.
        """
        resolved_labels = list(labels or [])
        resolved_titles = list(titles) if titles is not None else []
        row_count = cls._matrix_row_count(data)

        resolved_labels, resolved_titles = cls._resolve_row_alignment(
            row_count,
            resolved_labels,
            resolved_titles,
            "Label count must match the row count in the matrix-like data.",
            "Title count must match the row count in the matrix-like data.",
        )

        return cls(
            values=data,
            labels=resolved_labels,
            titles=resolved_titles,
            matrix=data,
            source="matrix",
        )

    @classmethod
    def _from_sequence_input(
        cls,
        data: Sequence[Any],
        labels: Sequence[Any] | None = None,
        titles: Sequence[Any] | None = None,
    ) -> "ClassifierData":
        """Normalize native Python sequences into a standardized Dataset wrapper.

        Args:
            data: The sequence of data items to normalize.
            labels: Optional sequence of labels to associate with the data. If not provided, labels will be inferred from the sequence.

        Returns:
            An instance of the ClassifierData class wrapping the normalized data.
        """
        resolved_labels = list(labels or [])
        resolved_titles = list(titles) if titles is not None else []

        resolved_labels, resolved_titles = cls._resolve_row_alignment(
            len(data),
            resolved_labels,
            resolved_titles,
            "Number of labels must match the number of data items.",
            "Title count must match the number of data items.",
        )

        return cls(
            values=list(data),
            labels=resolved_labels,
            titles=resolved_titles,
            source="raw",
        )

    @staticmethod
    def _resolve_row_alignment(
        row_count: int,
        labels: Sequence[Any] | None,
        titles: Sequence[Any] | None,
        labels_message: str,
        titles_message: str,
    ) -> tuple[list[Any], list[Any] | None]:
        """Validate label and title counts against a row count and return normalized values."""
        resolved_labels = list(labels or [])
        resolved_titles = list(titles) if titles is not None else []

        ClassifierData._validate_label_count(
            len(resolved_labels),
            row_count,
            labels_message,
        )
        if resolved_titles:
            ClassifierData._validate_label_count(
                len(resolved_titles),
                row_count,
                titles_message,
            )

        return resolved_labels, resolved_titles or None

    @classmethod
    def from_input(
        cls,
        data: Any,
        labels: Sequence[Any] | None = None,
        titles: Sequence[Any] | None = None,
    ) -> "ClassifierData":
        """Normalize raw text, DataFrames, and Lexos DTM objects to a standard form.

        Args:
            data: The input data to normalize. Can be raw text, a pandas DataFrame, a Lexos DTM object, or a matrix-like structure.
            labels: Optional sequence of labels to associate with the data. If not provided, labels will be inferred from the input data.
            titles: Optional sequence of titles aligned with the rows.

        Returns:
            An instance of the ClassifierData class wrapping the normalized data.
        """
        if _is_dtm_like(data):
            return cls._from_dtm_input(data, labels, titles)
        if isinstance(data, pd.DataFrame):
            return cls._from_dataframe_input(data, labels, titles)
        if hasattr(data, "shape") and getattr(data, "ndim", None) == 2:
            return cls._from_matrix_input(data, labels, titles)
        if isinstance(data, (list, tuple)):
            if data and _is_spacy_doc(data[0]):
                resolved_labels, resolved_titles = cls._resolve_row_alignment(
                    len(data),
                    labels,
                    titles,
                    "Number of labels must match the number of data items.",
                    "Title count must match the number of data items.",
                )
                return cls(
                    values=list(data),
                    labels=resolved_labels,
                    titles=resolved_titles,
                    docs=list(data),
                    source="raw",
                )
            return cls._from_sequence_input(data, labels, titles)
        if data is None:
            raise ValueError("No data was supplied to the classifier.")

        return cls(
            values=data,
            labels=list(labels or []),
            titles=list(titles) if titles is not None else None,
            source="raw",
        )

    def row_count(self) -> int:
        """Return the number of rows represented by the standardized input.

        Returns:
            The number of rows represented by the standardized input.
        """
        if self.matrix is not None:
            return self.matrix.shape[0]
        if self.docs is not None:
            return len(self.docs)
        if isinstance(self.values, pd.DataFrame):
            return len(self.values)
        if isinstance(self.values, (list, tuple)):
            return len(self.values)
        return len(self.labels)

    @staticmethod
    def _as_text_for_doc(doc: Any) -> str:
        """Convert a document-like object to text while preserving original tokenization.

        Args:
            doc: The document-like object to convert to text. Can be a spaCy Doc, an object with a `text` attribute, or a sequence of items.

        Returns:
            The text representation of the document-like object.
        """
        if doc is None:
            return ""

        module_name = type(doc).__module__ if doc is not None else ""
        if (
            module_name.startswith("spacy.")
            and hasattr(doc, "vocab")
            and hasattr(doc, "__iter__")
        ):
            return " ".join(token.text for token in doc)
        if hasattr(doc, "text") and not (
            hasattr(doc, "vocab") and hasattr(doc, "__iter__")
        ):
            return str(doc.text)
        if isinstance(doc, (list, tuple, set)):
            return " ".join(str(item) for item in doc)
        return str(doc)

    def as_texts(self) -> list[str]:
        """Return the data as plain text strings when possible.

        Returns:
            A list of plain text strings representing the data.
        """
        if self.docs is not None:
            texts: list[str] = []
            for doc in self.docs:
                texts.append(self._as_text_for_doc(doc))
            return texts

        if isinstance(self.values, pd.DataFrame):
            return [str(item) for item in self.values.to_dict(orient="records")]

        if isinstance(self.values, (list, tuple)):
            texts: list[str] = []
            for item in self.values:
                if _is_spacy_doc(item):
                    texts.append(" ".join(token.text for token in item))
                else:
                    texts.append(str(item))
            return texts

        return [str(self.values)]

    def subset(self, indices: Sequence[int]) -> "ClassifierData":
        """Return a new data object containing only the selected row indices.

        Args:
            indices: A sequence of row indices to include in the subset.

        Returns:
            A new `ClassifierData` object containing only the selected rows.
        """
        idx_list = list(indices)
        selected_titles = (
            [self.titles[i] for i in idx_list] if self.titles is not None else None
        )
        if self.matrix is not None:
            return ClassifierData(
                values=self.matrix[idx_list],
                labels=[self.labels[i] for i in idx_list],
                docs=[self.docs[i] for i in idx_list]
                if self.docs is not None
                else None,
                titles=selected_titles,
                matrix=self.matrix[idx_list],
                source=self.source,
            )

        if isinstance(self.values, pd.DataFrame):
            data = self.values.iloc[idx_list]
            return ClassifierData(
                values=data,
                labels=[self.labels[i] for i in idx_list],
                titles=selected_titles,
                source=self.source,
            )

        sliced = [self.values[i] for i in idx_list]
        return ClassifierData(
            values=sliced,
            labels=[self.labels[i] for i in idx_list],
            titles=selected_titles,
            source=self.source,
        )

    @staticmethod
    def _test_count_for_group(group_size: int, test_size: float) -> int:
        """Compute the number of rows to reserve for testing within one label group.

        Args:
            group_size: The number of rows in the label group.
            test_size: The proportion of the group to reserve for testing.

        Returns:
            The number of rows to reserve for testing within the group.
        """
        if group_size <= 1:
            return 0

        count = int(round(group_size * test_size))
        if count >= group_size:
            count = group_size - 1
        if count <= 0:
            count = 1
        return count

    @staticmethod
    def _dev_count_for_group(group_size: int, dev_size: float) -> int:
        """Compute the number of rows to reserve for development within a train split.

        Args:
            group_size: The number of rows in the train split.
            dev_size: The proportion of the train split to reserve for development.

        Returns:
            The number of rows to reserve for development within the train split.
        """
        if group_size <= 1:
            return 0

        count = int(round(group_size * dev_size))
        if count >= group_size:
            count = max(0, group_size - 1)
        if count <= 0:
            count = 1 if group_size > 1 else 0
        return count

    def _group_label_indices(self) -> dict[str, list[int]]:
        """Group row indices by label value for stratified splitting.

        Returns:
            A dictionary mapping each label value to a list of row indices that have that label.
        """
        grouped: dict[str, list[int]] = defaultdict(list)
        for idx, label in enumerate(self.labels):
            grouped[str(label)].append(idx)
        return grouped

    def _split_indices_by_labels(
        self,
        test_size: float,
        dev_size: float | None,
        random_state: int,
    ) -> dict[str, list[int]]:
        """Split row indices while preserving each class distribution.

        Args:
            test_size: The proportion of the dataset to reserve for testing.
            dev_size: The proportion of the training set to reserve for development, or None if no development set is needed.
            random_state: The seed for the random number generator to ensure reproducibility.

        Returns:
            A dictionary containing the split row indices with keys "train", "test", and optionally "dev".
        """
        grouped = self._group_label_indices()
        train_indices: list[int] = []
        test_indices: list[int] = []
        rng = random.Random(random_state)

        for label_indices in grouped.values():
            rng.shuffle(label_indices)
            label_test_count = self._test_count_for_group(len(label_indices), test_size)
            test_indices.extend(label_indices[:label_test_count])
            train_indices.extend(label_indices[label_test_count:])

        rng.shuffle(train_indices)
        rng.shuffle(test_indices)

        result = {"train": train_indices, "test": test_indices}
        if dev_size is not None:
            if len(train_indices) <= 1:
                result["dev"] = []
            else:
                dev_count = self._dev_count_for_group(len(train_indices), dev_size)
                dev_indices = train_indices[:dev_count]
                result["dev"] = dev_indices
                result["train"] = [
                    idx for idx in train_indices if idx not in set(dev_indices)
                ]
        return result

    @staticmethod
    def _validate_split_parameters(
        n_rows: int, test_size: float, dev_size: float | None
    ) -> None:
        """Validate dataset split parameters before partitioning.

        Args:
            n_rows: The total number of rows in the dataset.
            test_size: The proportion of the dataset to reserve for testing.
            dev_size: The proportion of the training set to reserve for development, or None if no development set is needed.

        Raises:
            ValueError: If any of the split parameters are invalid.
        """
        if n_rows == 0:
            raise ValueError("Cannot split an empty dataset.")
        if test_size <= 0 or test_size >= 1:
            raise ValueError("test_size must be between 0 and 1.")
        if dev_size is not None and (dev_size <= 0 or dev_size >= 1):
            raise ValueError("dev_size must be between 0 and 1.")

    @staticmethod
    def _split_random_indices(
        n_rows: int, test_size: float, random_state: int
    ) -> dict[str, list[int]]:
        """Create a simple non-stratified random split.

        Args:
            n_rows: The total number of rows in the dataset.
            test_size: The proportion of the dataset to reserve for testing.
            random_state: The seed for the random number generator to ensure reproducibility.

        Returns:
            A dictionary containing the split row indices with keys "train" and "test".
        """
        rng = random.Random(random_state)
        indices = list(range(n_rows))
        rng.shuffle(indices)

        test_count = int(round(n_rows * test_size))
        if n_rows > 1 and test_count >= n_rows:
            test_count = n_rows - 1
        if n_rows > 1 and test_count <= 0:
            test_count = 1
        if n_rows <= 1:
            test_count = 0

        test_indices = set(indices[:test_count])
        train_indices = [idx for idx in indices if idx not in test_indices]
        return {"train": train_indices, "test": list(test_indices)}

    def split(
        self,
        test_size: float = 0.2,
        dev_size: float | None = None,
        random_state: int = 42,
        stratify: bool = True,
    ) -> dict[str, "ClassifierData"]:
        """Split a standardized dataset into train/test/dev partitions.

        Args:
            test_size: The proportion of the dataset to reserve for testing.
            dev_size: The proportion of the training set to reserve for development, or None if no development set is needed.
            random_state: The seed for the random number generator to ensure reproducibility.
            stratify: Whether to perform a stratified split based on the labels.

        Returns:
            A dictionary containing the split datasets with keys "train", "test", and optionally "dev".
        """
        n_rows = self.row_count()
        self._validate_split_parameters(n_rows, test_size, dev_size)

        if stratify and self.labels:
            split = self._split_indices_by_labels(
                test_size=test_size,
                dev_size=dev_size,
                random_state=random_state,
            )
        else:
            split = self._split_random_indices(n_rows, test_size, random_state)

        train_result = self.subset(split["train"])
        test_result = self.subset(split["test"])

        result: dict[str, ClassifierData] = {
            "train": train_result,
            "test": test_result,
        }

        if dev_size is not None and "dev" in split:
            result["dev"] = self.subset(split["dev"])

        return result

__init__(values: Any, labels: Sequence[Any] | None = None, *, docs: Any = None, titles: Sequence[Any] | None = None, matrix: Any = None, source: str = 'raw') -> None

Initialize the ClassifierData object.

Parameters:

Name	Type	Description	Default
values	Any	The main data values.	required
labels	Sequence[Any] | None	Optional sequence of labels corresponding to the data.	None
docs	Any	Optional sequence of document objects.	None
titles	Sequence[Any] | None	Optional sequence of titles for the data items.	None
matrix	Any	Optional matrix representation of the data.	None
source	str	A string indicating the source of the data.	'raw'

Source code in lexos/classification/classifier.py

def __init__(
    self,
    values: Any,
    labels: Sequence[Any] | None = None,
    *,
    docs: Any = None,
    titles: Sequence[Any] | None = None,
    matrix: Any = None,
    source: str = "raw",
) -> None:
    """Initialize the ClassifierData object.

    Args:
        values: The main data values.
        labels: Optional sequence of labels corresponding to the data.
        docs: Optional sequence of document objects.
        titles: Optional sequence of titles for the data items.
        matrix: Optional matrix representation of the data.
        source: A string indicating the source of the data.
    """
    super().__init__(
        values=values,
        labels=list(labels) if labels is not None else [],
        docs=docs,
        titles=list(titles) if titles is not None else None,
        matrix=matrix,
        source=source,
    )

as_texts() -> list[str]

Return the data as plain text strings when possible.

Returns:

Type	Description
list[str]	A list of plain text strings representing the data.

Source code in lexos/classification/classifier.py

def as_texts(self) -> list[str]:
    """Return the data as plain text strings when possible.

    Returns:
        A list of plain text strings representing the data.
    """
    if self.docs is not None:
        texts: list[str] = []
        for doc in self.docs:
            texts.append(self._as_text_for_doc(doc))
        return texts

    if isinstance(self.values, pd.DataFrame):
        return [str(item) for item in self.values.to_dict(orient="records")]

    if isinstance(self.values, (list, tuple)):
        texts: list[str] = []
        for item in self.values:
            if _is_spacy_doc(item):
                texts.append(" ".join(token.text for token in item))
            else:
                texts.append(str(item))
        return texts

    return [str(self.values)]

from_input(data: Any, labels: Sequence[Any] | None = None, titles: Sequence[Any] | None = None) -> ClassifierData classmethod

Normalize raw text, DataFrames, and Lexos DTM objects to a standard form.

Parameters:

Name	Type	Description	Default
data	Any	The input data to normalize. Can be raw text, a pandas DataFrame, a Lexos DTM object, or a matrix-like structure.	required
labels	Sequence[Any] | None	Optional sequence of labels to associate with the data. If not provided, labels will be inferred from the input data.	None
titles	Sequence[Any] | None	Optional sequence of titles aligned with the rows.	None

Returns:

Type	Description
ClassifierData	An instance of the ClassifierData class wrapping the normalized data.

Source code in lexos/classification/classifier.py

@classmethod
def from_input(
    cls,
    data: Any,
    labels: Sequence[Any] | None = None,
    titles: Sequence[Any] | None = None,
) -> "ClassifierData":
    """Normalize raw text, DataFrames, and Lexos DTM objects to a standard form.

    Args:
        data: The input data to normalize. Can be raw text, a pandas DataFrame, a Lexos DTM object, or a matrix-like structure.
        labels: Optional sequence of labels to associate with the data. If not provided, labels will be inferred from the input data.
        titles: Optional sequence of titles aligned with the rows.

    Returns:
        An instance of the ClassifierData class wrapping the normalized data.
    """
    if _is_dtm_like(data):
        return cls._from_dtm_input(data, labels, titles)
    if isinstance(data, pd.DataFrame):
        return cls._from_dataframe_input(data, labels, titles)
    if hasattr(data, "shape") and getattr(data, "ndim", None) == 2:
        return cls._from_matrix_input(data, labels, titles)
    if isinstance(data, (list, tuple)):
        if data and _is_spacy_doc(data[0]):
            resolved_labels, resolved_titles = cls._resolve_row_alignment(
                len(data),
                labels,
                titles,
                "Number of labels must match the number of data items.",
                "Title count must match the number of data items.",
            )
            return cls(
                values=list(data),
                labels=resolved_labels,
                titles=resolved_titles,
                docs=list(data),
                source="raw",
            )
        return cls._from_sequence_input(data, labels, titles)
    if data is None:
        raise ValueError("No data was supplied to the classifier.")

    return cls(
        values=data,
        labels=list(labels or []),
        titles=list(titles) if titles is not None else None,
        source="raw",
    )

row_count() -> int

Return the number of rows represented by the standardized input.

Returns:

Type	Description
int	The number of rows represented by the standardized input.

Source code in lexos/classification/classifier.py

def row_count(self) -> int:
    """Return the number of rows represented by the standardized input.

    Returns:
        The number of rows represented by the standardized input.
    """
    if self.matrix is not None:
        return self.matrix.shape[0]
    if self.docs is not None:
        return len(self.docs)
    if isinstance(self.values, pd.DataFrame):
        return len(self.values)
    if isinstance(self.values, (list, tuple)):
        return len(self.values)
    return len(self.labels)

split(test_size: float = 0.2, dev_size: float | None = None, random_state: int = 42, stratify: bool = True) -> dict[str, ClassifierData]

Split a standardized dataset into train/test/dev partitions.

Parameters:

Name	Type	Description	Default
test_size	float	The proportion of the dataset to reserve for testing.	0.2
dev_size	float | None	The proportion of the training set to reserve for development, or None if no development set is needed.	None
random_state	int	The seed for the random number generator to ensure reproducibility.	42
stratify	bool	Whether to perform a stratified split based on the labels.	True

Returns:

Type	Description
dict[str, ClassifierData]	A dictionary containing the split datasets with keys "train", "test", and optionally "dev".

Source code in lexos/classification/classifier.py

def split(
    self,
    test_size: float = 0.2,
    dev_size: float | None = None,
    random_state: int = 42,
    stratify: bool = True,
) -> dict[str, "ClassifierData"]:
    """Split a standardized dataset into train/test/dev partitions.

    Args:
        test_size: The proportion of the dataset to reserve for testing.
        dev_size: The proportion of the training set to reserve for development, or None if no development set is needed.
        random_state: The seed for the random number generator to ensure reproducibility.
        stratify: Whether to perform a stratified split based on the labels.

    Returns:
        A dictionary containing the split datasets with keys "train", "test", and optionally "dev".
    """
    n_rows = self.row_count()
    self._validate_split_parameters(n_rows, test_size, dev_size)

    if stratify and self.labels:
        split = self._split_indices_by_labels(
            test_size=test_size,
            dev_size=dev_size,
            random_state=random_state,
        )
    else:
        split = self._split_random_indices(n_rows, test_size, random_state)

    train_result = self.subset(split["train"])
    test_result = self.subset(split["test"])

    result: dict[str, ClassifierData] = {
        "train": train_result,
        "test": test_result,
    }

    if dev_size is not None and "dev" in split:
        result["dev"] = self.subset(split["dev"])

    return result

subset(indices: Sequence[int]) -> ClassifierData

Return a new data object containing only the selected row indices.

Parameters:

Name	Type	Description	Default
indices	Sequence[int]	A sequence of row indices to include in the subset.	required

Returns:

Type	Description
ClassifierData	A new ClassifierData object containing only the selected rows.

Source code in lexos/classification/classifier.py

def subset(self, indices: Sequence[int]) -> "ClassifierData":
    """Return a new data object containing only the selected row indices.

    Args:
        indices: A sequence of row indices to include in the subset.

    Returns:
        A new `ClassifierData` object containing only the selected rows.
    """
    idx_list = list(indices)
    selected_titles = (
        [self.titles[i] for i in idx_list] if self.titles is not None else None
    )
    if self.matrix is not None:
        return ClassifierData(
            values=self.matrix[idx_list],
            labels=[self.labels[i] for i in idx_list],
            docs=[self.docs[i] for i in idx_list]
            if self.docs is not None
            else None,
            titles=selected_titles,
            matrix=self.matrix[idx_list],
            source=self.source,
        )

    if isinstance(self.values, pd.DataFrame):
        data = self.values.iloc[idx_list]
        return ClassifierData(
            values=data,
            labels=[self.labels[i] for i in idx_list],
            titles=selected_titles,
            source=self.source,
        )

    sliced = [self.values[i] for i in idx_list]
    return ClassifierData(
        values=sliced,
        labels=[self.labels[i] for i in idx_list],
        titles=selected_titles,
        source=self.source,
    )

rendering:
  show_root_heading: true
  heading_level: 3