Pipeline

Meta-estimators that combine branches, pick fields, work on every item of split samples, and retrieve items.

Field(name)

Bases: TransformerMixin, BaseEstimator

Return one field of every multimodal sample or region, or a dict of several fields.

It is stateless, so it can sit anywhere in a pipeline, for example after an extractor's predict or inside ForEach to pick the part of every region that the next step needs.

Parameters:

Name	Type	Description	Default
name	str | list[str]	The name of the field, or a list of names to return dicts of those fields.	required

Examples:

>>> from skai.pipeline import Field
>>> Field('visual').fit_transform([{'visual': 'a boat', 'narration': 'At sea.'}]).tolist()
['a boat']

Source code in src/skai/pipeline/_field.py

def __init__(self, name: str | list[str]) -> None:
    self.name = name

__sklearn_tags__()

Declare that the transformer is stateless.

Source code in src/skai/pipeline/_field.py

def __sklearn_tags__(self) -> Tags:
    """Declare that the transformer is stateless."""
    tags = super().__sklearn_tags__()
    tags.requires_fit = False
    tags.input_tags.two_d_array = False
    return tags

fit(X=None, y=None)

Validate the parameters.

Parameters:

Name	Type	Description	Default
X	ArrayLike | None	Ignored.	None
y	ArrayLike | None	Ignored.	None

Returns:

Type	Description
Self	The transformer.

Source code in src/skai/pipeline/_field.py

def fit(self, X: ArrayLike | None = None, y: ArrayLike | None = None) -> Self:
    """Validate the parameters.

    Args:
        X:
            Ignored.

        y:
            Ignored.

    Returns:
        The transformer.
    """
    self._validate_params()
    return self

transform(X)

Pick the field of every sample.

Parameters:

Name	Type	Description	Default
X	ArrayLike	Dicts, such as multimodal samples, extracted records or regions.	required

Returns:

Type	Description
NDArray[object_]	The value of the field per sample, or a dict of the fields when name is a list.

Raises:

Type	Description
KeyError	If a sample lacks a field.

Source code in src/skai/pipeline/_field.py

def transform(self, X: ArrayLike) -> NDArray[np.object_]:
    """Pick the field of every sample.

    Args:
        X:
            Dicts, such as multimodal samples, extracted records or regions.

    Returns:
        The value of the field per sample, or a dict of the fields when `name` is a list.

    Raises:
        KeyError:
            If a sample lacks a field.
    """
    names = self.name if isinstance(self.name, list) else [self.name]
    values = []
    for index, sample in enumerate(to_samples(X, rows=True)):
        missing = [name for name in names if not isinstance(sample, dict) or name not in sample]
        if missing:
            msg = f'Sample {index} has no field `{missing[0]}`.'
            raise KeyError(msg)
        picked: Any = {name: sample[name] for name in names} if isinstance(self.name, list) else sample[self.name]
        values.append(picked)
    return to_object_array(values)

ForEach(estimator)

Bases: TransformerMixin, BaseEstimator

Apply an estimator to every item of list-valued samples, such as the regions of a segmenter.

The items of all samples are flattened and processed in one call of the estimator, so a backend receives them as one batch, and the outputs are split back into lists of the original lengths. fit fits the estimator on all the items with y=None.

Parameters:

Name	Type	Description	Default
estimator	BaseEstimator	The estimator or pipeline applied to every item.	required

Attributes:

Name	Type	Description
estimator_	BaseEstimator	The fitted estimator.

Examples:

>>> from skai.generation import TextToTextGenerator
>>> from skai.pipeline import ForEach
>>> generator = TextToTextGenerator(instructions='Shorten.', max_length=1, backend='dummy')
>>> ForEach(generator).fit_transform([['One two', 'Three four'], []]).tolist()
[['One', 'Three'], []]

Source code in src/skai/pipeline/_for_each.py

def __init__(self, estimator: BaseEstimator) -> None:
    self.estimator = estimator

__sklearn_tags__()

Declare that the samples are lists.

Source code in src/skai/pipeline/_for_each.py

def __sklearn_tags__(self) -> Tags:
    """Declare that the samples are lists."""
    tags = super().__sklearn_tags__()
    tags.input_tags.two_d_array = False
    return tags

fit(X, y=None)

Fit the estimator on the items of all samples.

Parameters:

Name	Type	Description	Default
X	ArrayLike	The list-valued samples.	required
y	ArrayLike | None	Ignored, since items have no targets.	None

Returns:

Type	Description
Self	The fitted transformer.

Source code in src/skai/pipeline/_for_each.py

def fit(self, X: ArrayLike, y: ArrayLike | None = None) -> Self:
    """Fit the estimator on the items of all samples.

    Args:
        X:
            The list-valued samples.

        y:
            Ignored, since items have no targets.

    Returns:
        The fitted transformer.
    """
    items, _ = _flatten(X)
    self.estimator_ = clone(self.estimator).fit(items or None)
    return self

fit_transform(X, y=None, **fit_params)

Fit the estimator on the items of all samples and return their outputs, running every item once.

Parameters:

Name	Type	Description	Default
X	ArrayLike	The list-valued samples.	required
y	ArrayLike | None	Ignored, since items have no targets.	None
**fit_params	Any	Ignored.	{}

Returns:

Type	Description
NDArray[object_]	One list of outputs per sample.

Source code in src/skai/pipeline/_for_each.py

def fit_transform(self, X: ArrayLike, y: ArrayLike | None = None, **fit_params: Any) -> NDArray[np.object_]:
    """Fit the estimator on the items of all samples and return their outputs, running every item once.

    Args:
        X:
            The list-valued samples.

        y:
            Ignored, since items have no targets.

        **fit_params:
            Ignored.

    Returns:
        One list of outputs per sample.
    """
    items, lengths = _flatten(X)
    self.estimator_ = clone(self.estimator)
    if not items:
        self.estimator_.fit(None)
        return to_object_array([] for _ in lengths)
    estimator = self.estimator_
    output = estimator.fit_transform(items) if hasattr(estimator, 'fit_transform') else None
    if output is None:
        output = estimator.fit(items).predict(items)
    return _split(output, lengths)

predict(X)

Predict on the items of all samples.

Parameters:

Name	Type	Description	Default
X	ArrayLike	The list-valued samples.	required

Returns:

Type	Description
NDArray[object_]	One list of predictions per sample, or of outputs when the estimator only transforms.

Source code in src/skai/pipeline/_for_each.py

def predict(self, X: ArrayLike) -> NDArray[np.object_]:
    """Predict on the items of all samples.

    Args:
        X:
            The list-valued samples.

    Returns:
        One list of predictions per sample, or of outputs when the estimator only transforms.
    """
    check_is_fitted(self, 'estimator_')
    return self._apply(X, 'predict' if hasattr(self.estimator_, 'predict') else 'transform')

transform(X)

Transform the items of all samples.

Parameters:

Name	Type	Description	Default
X	ArrayLike	The list-valued samples.	required

Returns:

Type	Description
NDArray[object_]	One list of outputs per sample.

Source code in src/skai/pipeline/_for_each.py

def transform(self, X: ArrayLike) -> NDArray[np.object_]:
    """Transform the items of all samples.

    Args:
        X:
            The list-valued samples.

    Returns:
        One list of outputs per sample.
    """
    return self._apply(X, 'transform' if hasattr(self.estimator_, 'transform') else 'predict')

MediaUnion(transformer_list)

Bases: TransformerMixin, _BaseComposition

Run named branches on the same samples and return one multimodal sample, a dict keyed by branch name, per sample.

It is modeled on FeatureUnion: every branch is an estimator, a pipeline or 'passthrough', branches are fitted through fit_transform in pipelines, and their parameters are routed as <branch>__<parameter>. A branch that cannot transform, such as one ending in a classifier, contributes its predictions. Instead of concatenating columns, it keeps the output of every branch under its name, so later steps such as MultimodalToVideoGenerator receive several inputs per sample.

Parameters:

Name	Type	Description	Default
transformer_list	list[tuple[str, BaseEstimator | str]]	The branches, as (name, estimator) pairs. An estimator can be 'passthrough' to keep the input.	required

Attributes:

Name	Type	Description
transformers_	list	The fitted branches, as (name, estimator) pairs.

Examples:

>>> from skai.generation import TextToTextGenerator
>>> from skai.pipeline import MediaUnion
>>> summary = TextToTextGenerator(instructions='Summarize.', max_length=1, backend='dummy')
>>> union = MediaUnion([('text', 'passthrough'), ('summary', summary)])
>>> union.fit_transform(['Hello world'])[0]
{'text': 'Hello world', 'summary': 'Hello'}

Source code in src/skai/pipeline/_union.py

def __init__(self, transformer_list: list[tuple[str, BaseEstimator | str]]) -> None:
    self.transformer_list = transformer_list

fit(X, y=None)

Fit every branch on the samples.

Parameters:

Name	Type	Description	Default
X	ArrayLike	The samples.	required
y	ArrayLike | None	The targets, passed to every branch.	None

Returns:

Type	Description
Self	The fitted union.

Source code in src/skai/pipeline/_union.py

def fit(self, X: ArrayLike, y: ArrayLike | None = None) -> Self:
    """Fit every branch on the samples.

    Args:
        X:
            The samples.

        y:
            The targets, passed to every branch.

    Returns:
        The fitted union.
    """
    self._check()
    self.transformers_ = [
        (name, branch if branch == PASSTHROUGH else clone(branch).fit(X, y))
        for name, branch in self.transformer_list
    ]
    return self

fit_transform(X, y=None, **fit_params)

Fit every branch through its fit_transform and return the multimodal samples.

Parameters:

Name	Type	Description	Default
X	ArrayLike	The samples.	required
y	ArrayLike | None	The targets, passed to every branch.	None
**fit_params	Any	Ignored.	{}

Returns:

Type	Description
NDArray[object_]	One dict per sample, keyed by branch name.

Source code in src/skai/pipeline/_union.py

def fit_transform(self, X: ArrayLike, y: ArrayLike | None = None, **fit_params: Any) -> NDArray[np.object_]:
    """Fit every branch through its `fit_transform` and return the multimodal samples.

    Args:
        X:
            The samples.

        y:
            The targets, passed to every branch.

        **fit_params:
            Ignored.

    Returns:
        One dict per sample, keyed by branch name.
    """
    self._check()
    samples = to_samples(X)
    self.transformers_ = []
    outputs = {}
    for name, branch in self.transformer_list:
        if branch == PASSTHROUGH:
            self.transformers_.append((name, branch))
            outputs[name] = samples
            continue
        fitted = clone(branch)
        output = fitted.fit_transform(X, y) if hasattr(fitted, 'transform') else fitted.fit(X, y).predict(X)
        outputs[name] = _rows(output, len(samples))
        self.transformers_.append((name, fitted))
    return self._join(outputs, len(samples))

get_params(deep=True)

Return the parameters of the union and, with deep, of its branches.

Parameters:

Name	Type	Description	Default
deep	bool	Whether to include the parameters of the branches as <branch>__<parameter>.	True

Returns:

Type	Description
dict[str, Any]	The parameters.

Source code in src/skai/pipeline/_union.py

def get_params(self, deep: bool = True) -> dict[str, Any]:
    """Return the parameters of the union and, with `deep`, of its branches.

    Args:
        deep:
            Whether to include the parameters of the branches as `<branch>__<parameter>`.

    Returns:
        The parameters.
    """
    return self._get_params('transformer_list', deep=deep)

predict(X)

Run every branch and return the multimodal samples, so the union can be the final step of a pipeline.

Parameters:

Name	Type	Description	Default
X	ArrayLike	The samples.	required

Returns:

Type	Description
NDArray[object_]	One dict per sample, keyed by branch name.

Source code in src/skai/pipeline/_union.py

def predict(self, X: ArrayLike) -> NDArray[np.object_]:
    """Run every branch and return the multimodal samples, so the union can be the final step of a pipeline.

    Args:
        X:
            The samples.

    Returns:
        One dict per sample, keyed by branch name.
    """
    return self.transform(X)

set_params(**params)

Set the parameters of the union or of its branches.

Parameters:

Name	Type	Description	Default
**params	Any	The parameters, with <branch> to replace a branch and <branch>__<parameter> for its parameters.	{}

Returns:

Type	Description
Self	The union.

Source code in src/skai/pipeline/_union.py

def set_params(self, **params: Any) -> Self:
    """Set the parameters of the union or of its branches.

    Args:
        **params:
            The parameters, with `<branch>` to replace a branch and `<branch>__<parameter>` for its parameters.

    Returns:
        The union.
    """
    self._set_params('transformer_list', **params)
    return self

transform(X)

Run every branch and return the multimodal samples.

Parameters:

Name	Type	Description	Default
X	ArrayLike	The samples.	required

Returns:

Type	Description
NDArray[object_]	One dict per sample, keyed by branch name.

Source code in src/skai/pipeline/_union.py

def transform(self, X: ArrayLike) -> NDArray[np.object_]:
    """Run every branch and return the multimodal samples.

    Args:
        X:
            The samples.

    Returns:
        One dict per sample, keyed by branch name.
    """
    check_is_fitted(self, 'transformers_')
    samples = to_samples(X)
    outputs = {
        name: samples if branch == PASSTHROUGH else _rows(_apply(branch, X), len(samples))
        for name, branch in self.transformers_
    }
    return self._join(outputs, len(samples))

Retriever(items, queries, *, metric='cosine')

Bases: BaseEstimator

Index the items of samples with an item pipeline and find the nearest items of queries.

The item pipeline returns, per sample, either one vector, or a list of records with a vector key, such as the shots of a video with their times and embeddings. fit indexes every item with the index of its sample, and kneighbors embeds queries with the query pipeline, which is fitted without data, and returns the records of the nearest items. With a joint text and image model, texts find images, shots or videos.

Parameters:

Name	Type	Description	Default
items	BaseEstimator	The pipeline that turns every sample into a vector or a list of records with vectors.	required
queries	BaseEstimator	The pipeline that turns every query into a vector.	required
metric	str	The distance between vectors.	'cosine'

Attributes:

Name	Type	Description
items_	BaseEstimator	The fitted item pipeline.
queries_	BaseEstimator	The fitted query pipeline.
records_	list	The records of the indexed items, without their vectors.
index_	NearestNeighbors	The index of the item vectors.

Examples:

>>> from skai.embedding import TextEmbedder
>>> from skai.pipeline import Retriever
>>> retriever = Retriever(TextEmbedder(center=False, backend='dummy'), TextEmbedder(backend='dummy'))
>>> retriever.fit(['a goal', 'a corner', 'a foul']).kneighbors(['a goal'], n_neighbors=1)[0][0]['sample']
0

Source code in src/skai/pipeline/_retriever.py

def __init__(self, items: BaseEstimator, queries: BaseEstimator, *, metric: str = 'cosine') -> None:
    self.items = items
    self.queries = queries
    self.metric = metric

__sklearn_tags__()

Declare that samples are media, not a numeric matrix.

Source code in src/skai/pipeline/_retriever.py

def __sklearn_tags__(self) -> Tags:
    """Declare that samples are media, not a numeric matrix."""
    tags = super().__sklearn_tags__()
    tags.input_tags.two_d_array = False
    return tags

fit(X, y=None)

Index the items of the samples.

Parameters:

Name	Type	Description	Default
X	ArrayLike	The samples, such as videos.	required
y	ArrayLike | None	Ignored.	None

Returns:

Type	Description
Self	The fitted retriever.

Raises:

Type	Description
ValueError	If the samples have no items.

Source code in src/skai/pipeline/_retriever.py

def fit(self, X: ArrayLike, y: ArrayLike | None = None) -> Self:
    """Index the items of the samples.

    Args:
        X:
            The samples, such as videos.

        y:
            Ignored.

    Returns:
        The fitted retriever.

    Raises:
        ValueError:
            If the samples have no items.
    """
    self._validate_params()
    self.items_ = clone(self.items)
    records = _records(self.items_.fit_transform(X))
    if not records:
        msg = 'The samples have no items to index.'
        raise ValueError(msg)
    vectors = np.stack([np.asarray(record.pop(VECTOR), dtype=float) for record in records])
    self.records_ = records
    self.index_ = NearestNeighbors(metric=self.metric).fit(vectors)
    self.queries_ = clone(self.queries).fit(None)
    return self

kneighbors(queries, n_neighbors=5)

Find the nearest items of every query.

Parameters:

Name	Type	Description	Default
queries	ArrayLike	The queries, such as texts.	required
n_neighbors	int	The number of items per query, at most the number of indexed items.	5

Returns:

Type	Description
NDArray[object_]	One list of item records per query, nearest first, each with its sample index and distance.

Source code in src/skai/pipeline/_retriever.py

def kneighbors(self, queries: ArrayLike, n_neighbors: int = 5) -> NDArray[np.object_]:
    """Find the nearest items of every query.

    Args:
        queries:
            The queries, such as texts.

        n_neighbors:
            The number of items per query, at most the number of indexed items.

    Returns:
        One list of item records per query, nearest first, each with its `sample` index and `distance`.
    """
    check_is_fitted(self, 'index_')
    vectors = np.asarray(self.queries_.transform(queries), dtype=float)
    count = min(n_neighbors, len(self.records_))
    distances, indices = self.index_.kneighbors(vectors, n_neighbors=count)
    return to_object_array(
        [{**self.records_[index], 'distance': float(distance)} for index, distance in zip(row, gaps, strict=True)]
        for row, gaps in zip(indices, distances, strict=True)
    )

TopK(scorer, *, label=None, k=None, threshold=None)

Bases: TransformerMixin, BaseEstimator

Keep the k items of every list-valued sample with the highest scores, or those above threshold.

The kept items stay in their original order. A score comes from a callable on one item, or from the predict_proba column of label of a classifier, which fit fits zero-shot on the items. The items of all samples are scored in one call.

Parameters:

Name	Type	Description	Default
scorer	Callable[[Any], float] | BaseEstimator	A callable that scores one item, or a classifier with predict_proba, such as a VideoClassifier with labels.	required
label	str | int | None	The class whose probability is the score of a classifier.	None
k	int | None	The number of items to keep per sample. All items above threshold are kept when it is None.	None
threshold	float | None	The minimum score of a kept item.	None

Attributes:

Name	Type	Description
scorer_	Callable | BaseEstimator	The callable, or the fitted classifier.

Examples:

>>> from skai.pipeline import TopK
>>> TopK(len, k=2).fit_transform([['a', 'abc', 'ab', 'abcd']]).tolist()
[['abc', 'abcd']]

Source code in src/skai/pipeline/_top_k.py

def __init__(
    self,
    scorer: Callable[[Any], float] | BaseEstimator,
    *,
    label: str | int | None = None,
    k: int | None = None,
    threshold: float | None = None,
) -> None:
    self.scorer = scorer
    self.label = label
    self.k = k
    self.threshold = threshold

__sklearn_tags__()

Declare that the samples are lists.

Source code in src/skai/pipeline/_top_k.py

def __sklearn_tags__(self) -> Tags:
    """Declare that the samples are lists."""
    tags = super().__sklearn_tags__()
    tags.input_tags.two_d_array = False
    return tags

fit(X=None, y=None)

Fit a classifier scorer zero-shot, without targets.

Parameters:

Name	Type	Description	Default
X	ArrayLike | None	The list-valued samples.	None
y	ArrayLike | None	Ignored, since items have no targets.	None

Returns:

Type	Description
Self	The fitted transformer.

Raises:

Type	Description
ValueError	If neither k nor threshold is given, or a classifier is given without label.

Source code in src/skai/pipeline/_top_k.py

def fit(self, X: ArrayLike | None = None, y: ArrayLike | None = None) -> Self:
    """Fit a classifier scorer zero-shot, without targets.

    Args:
        X:
            The list-valued samples.

        y:
            Ignored, since items have no targets.

    Returns:
        The fitted transformer.

    Raises:
        ValueError:
            If neither `k` nor `threshold` is given, or a classifier is given without `label`.
    """
    self._validate_params()
    if self.k is None and self.threshold is None:
        msg = 'TopK needs `k` or `threshold`.'
        raise ValueError(msg)
    if hasattr(self.scorer, 'predict_proba'):
        if self.label is None:
            msg = 'TopK needs `label` to score items with a classifier.'
            raise ValueError(msg)
        self.scorer_ = clone(self.scorer).fit()
    else:
        self.scorer_ = self.scorer
    return self

transform(X)

Keep the best items of every sample.

Parameters:

Name	Type	Description	Default
X	ArrayLike	The list-valued samples.	required

Returns:

Type	Description
NDArray[object_]	One list of kept items per sample, in their original order.

Source code in src/skai/pipeline/_top_k.py

def transform(self, X: ArrayLike) -> NDArray[np.object_]:
    """Keep the best items of every sample.

    Args:
        X:
            The list-valued samples.

    Returns:
        One list of kept items per sample, in their original order.
    """
    check_is_fitted(self, 'scorer_')
    items, lengths = _flatten(X)
    scores = self._scores(items) if items else np.zeros(0)
    kept = []
    start = 0
    for length in lengths:
        indices = np.arange(start, start + length)
        if self.threshold is not None:
            indices = indices[scores[indices] > self.threshold]
        if self.k is not None:
            indices = np.sort(indices[np.argsort(-scores[indices], kind='stable')[: self.k]])
        kept.append([items[index] for index in indices])
        start += length
    return to_object_array(kept)