Classification

The Classification task categorizes documents into predefined labels.

Usage

Simple List of Labels

...

task = tasks.Classification(
    labels={
        "positive": "Positive sentiment regarding the subject.",
        "negative": "Negative sentiment regarding the subject.",
    },
    model=model,
)

Results

The Classification task returns a unified result schema regardless of the model backend used.

class ResultSingleLabel(pydantic.BaseModel):
    """Result of a single-label classification task.

    Attributes:
        label: Predicted label.
        score: Confidence score.
    """

    label: str
    score: float


class ResultMultiLabel(pydantic.BaseModel):
    """Result of a multi-label classification task.

    Attributes:
        label_scores: List of label-score pairs.
    """

    label_scores: list[tuple[str, float]]

• When mode == 'multi' (default): results are of type ResultMultiLabel, containing a list of (label, score) tuples.
• When mode == 'single': results are of type ResultSingleLabel, containing a single label and score.

Confidence scores are always present for transformers and gliner2 models. For LLMs, scores are self-reported and may be None.

Evaluation

You can evaluate the performance of your classifier using the .evaluate() method.

• Metric: Macro-averaged F1 Score (F1 (Macro)). This is calculated corpus-wide using scikit-learn.
• Requirement: Each document must have its ground-truth label stored in doc.gold[task_id].

Ground Truth Formats

For convenience, you can provide ground-truth data in simplified formats:

• Single-label (str): Just the label string.

  doc.gold["clf"] = "science"
  

• Multi-label (list[str]): A list of active labels.

  doc.gold["clf"] = ["science", "politics"]
  

Alternatively, you can use the standard Pydantic result objects (ResultSingleLabel, ResultMultiLabel) if you need to specify confidence scores for soft evaluation (though F1 uses hard labels).

report = task.evaluate(docs)
print(f"Classification Score: {report.metrics['F1 (Macro)']}")

---

Classification predictive task and few‑shot example schemas.

Classification

Bases: PredictiveTask[TaskPromptSignature, TaskResult, _TaskBridge]

Predictive task for text classification.

Source code in sieves/tasks/predictive/classification/core.py

class Classification(PredictiveTask[TaskPromptSignature, TaskResult, _TaskBridge]):
    """Predictive task for text classification."""

    def __init__(
        self,
        labels: Sequence[str] | dict[str, str],
        model: TaskModel,
        task_id: str | None = None,
        include_meta: bool = True,
        batch_size: int = -1,
        prompt_instructions: str | None = None,
        fewshot_examples: Sequence[FewshotExample] = (),
        mode: Literal["single", "multi"] = "multi",
        model_settings: ModelSettings = ModelSettings(),
        condition: Callable[[Doc], bool] | None = None,
    ) -> None:
        """Initialize new Classification task.

        :param labels: Labels to predict. Supports two formats:
            - List format: `["science", "politics", "sports"]`
            - Dict format: `{"science": "Scientific topics", "politics": "Political topics"}`
            The dict format allows you to provide descriptions that help the model better understand each label.
        :param model: Model to use.
        :param task_id: Task ID.
        :param include_meta: Whether to include meta information generated by the task.
        :param batch_size: Batch size to use for inference. Use -1 to process all documents at once.
        :param prompt_instructions: Custom prompt instructions. If None, default instructions are used.
        :param fewshot_examples: Few-shot examples.
        :param mode: If 'multi', task returns confidence scores for all specified labels. If 'single', task returns
            most likely class label.
        :param model_settings: Model settings.
        :param condition: Optional callable that determines whether to process each document.
        """
        if isinstance(labels, dict):
            self._labels = list(labels.keys())
            self._label_descriptions = labels
        else:
            self._labels = list(labels)
            self._label_descriptions = {}
        self._mode = mode

        super().__init__(
            model=model,
            task_id=task_id,
            include_meta=include_meta,
            batch_size=batch_size,
            overwrite=False,
            prompt_instructions=prompt_instructions,
            fewshot_examples=fewshot_examples,
            model_settings=model_settings,
            condition=condition,
        )

    @property
    @override
    def fewshot_example_type(self) -> type[FewshotExample]:
        """Return few-shot example type.

        :return: Few-shot example type.
        """
        if self._mode == "multi":
            return FewshotExampleMultiLabel

        return FewshotExampleSingleLabel

    @property
    @override
    def prompt_signature(self) -> type[pydantic.BaseModel]:
        if self._mode == "single":
            labels = self._labels
            LabelType = Literal[*labels]  # type: ignore[valid-type]

            class SingleLabelClassification(pydantic.BaseModel):
                """Result of single-label classification. Contains the most likely label and its confidence score."""

                label: LabelType = pydantic.Field(description="The predicted label from the specified set of labels.")
                score: float = pydantic.Field(
                    default=None, description="Provide a confidence score for the predicted label, between 0 and 1."
                )

            return SingleLabelClassification

        # For multi-label, create a model with fields for each label.
        fields: dict[str, tuple[type, Any]] = {}
        for label in self._labels:
            assert isinstance(self._label_descriptions, dict)
            description = self._label_descriptions.get(
                label, f"Confidence score for the '{label}' category, between 0 and 1."
            )  # type: ignore[no-matching-overload]
            fields[label] = (float, pydantic.Field(description=description))

        return pydantic.create_model(  # type: ignore[no-matching-overload]
            "MultiLabelClassification",
            __base__=pydantic.BaseModel,
            __doc__="Result of multi-label classification. Contains confidence scores for each potential label.",
            **fields,
        )

    @property
    @override
    def metric(self) -> str:
        return "F1 (Macro)"

    @override
    def _compute_metrics(self, truths: list[Any], preds: list[Any], judge: dspy.LM | None = None) -> dict[str, float]:
        """Compute corpus-level metrics.

        :param truths: List of ground truths.
        :param preds: List of predictions.
        :param judge: Optional DSPy LM instance to use as judge for generative tasks.
        :return: Dictionary of metrics.
        """
        # Prepare labels and mappings.
        labels_list = list(self._labels)
        label_to_idx = {label: i for i, label in enumerate(labels_list)}

        y_true: list[float | list[float] | int] = []
        y_pred: list[float | list[float] | int] = []

        for gold, pred in zip(truths, preds):
            if gold is None or pred is None:
                # If either is None, we handle it as a failure for this example.
                # In multi-label, this is an all-zero vector. In single-label, it's -1 (no class).
                if self._mode == "multi":
                    y_true.append([0.0] * len(labels_list))
                    y_pred.append([0.0] * len(labels_list))
                else:
                    y_true.append(-1)
                    y_pred.append(-1)
                continue

            # Convert to normalized representation using the existing logic.
            # We reuse `_task_result_to_dspy_dict` indirectly or just use `_result_to_scores`.
            gold_scores = self._result_to_scores(self._task_result_to_pydantic(gold))
            pred_scores = self._result_to_scores(self._task_result_to_pydantic(pred))

            if self._mode == "multi":
                # Binary multi-hot vectors.
                y_true.append([1.0 if gold_scores.get(label, 0.0) >= self.THRESHOLD else 0.0 for label in labels_list])
                y_pred.append([1.0 if pred_scores.get(label, 0.0) >= self.THRESHOLD else 0.0 for label in labels_list])
            else:
                # Single label indices.
                # Get the label with the highest score.
                gold_label = max(gold_scores.items(), key=lambda x: x[1])[0]
                pred_label = max(pred_scores.items(), key=lambda x: x[1])[0]
                y_true.append(label_to_idx.get(gold_label, -1))
                y_pred.append(label_to_idx.get(pred_label, -1))

        # Filter out examples where gold or pred were None (represented by -1 or all-zero if that's what we chose).
        score = sklearn.metrics.f1_score(y_true, y_pred, average="macro", zero_division=0)

        return {self.metric: float(score)}

    def _init_bridge(self, model_type: ModelType) -> _TaskBridge:
        """Initialize bridge.

        :return: ModelWrapper task.
        :raises ValueError: If model type is not supported.
        """
        # Reconstruct labels parameter (as dict if descriptions exist, else as list)
        labels = self._label_descriptions if self._label_descriptions else self._labels

        if model_type == ModelType.gliner:
            return GliNERBridge(
                task_id=self._task_id,
                prompt_instructions=self._custom_prompt_instructions,
                prompt_signature=self.prompt_signature,
                model_settings=self._model_settings,
                inference_mode=gliner_.InferenceMode.classification,
                mode=self._mode,
            )

        bridge_types: dict[ModelType, type[_TaskBridge]] = {
            ModelType.dspy: DSPyClassification,
            ModelType.huggingface: HuggingFaceClassification,
            ModelType.outlines: OutlinesClassification,
            ModelType.langchain: LangChainClassification,
        }

        try:
            bridge_type = bridge_types[model_type]
            assert not issubclass(bridge_type, GliNERBridge)

            return bridge_type(
                task_id=self._task_id,
                prompt_instructions=self._custom_prompt_instructions,
                labels=labels,
                mode=self._mode,
                model_settings=self._model_settings,
                prompt_signature=self.prompt_signature,
                model_type=model_type,
                fewshot_examples=self._fewshot_examples,
            )
        except KeyError as err:
            raise KeyError(f"Model type {model_type} is not supported by {self.__class__.__name__}.") from err

    @staticmethod
    @override
    def supports() -> set[ModelType]:
        return {
            ModelType.dspy,
            ModelType.gliner,
            ModelType.huggingface,
            ModelType.langchain,
            ModelType.outlines,
        }

    def _validate_fewshot_examples(self) -> None:
        label_error_text = (
            "Label mismatch: {task_id} has labels {labels}. Few-shot examples have labels {example_labels}."
        )
        example_type_error_text = "Fewshot example type mismatch: mode = {mode} requires {example_type}."

        for fs_example in self._fewshot_examples or []:
            if self._mode == "multi":
                assert isinstance(fs_example, FewshotExampleMultiLabel), TypeError(
                    example_type_error_text.format(example_type=FewshotExampleMultiLabel, mode=self._mode)
                )
                if any([label not in self._labels for label in fs_example.score_per_label]) or not all(
                    [label in fs_example.score_per_label for label in self._labels]
                ):
                    raise ValueError(
                        label_error_text.format(
                            task_id=self.id, labels=self._labels, example_labels=fs_example.score_per_label.keys()
                        )
                    )
            else:
                assert isinstance(fs_example, FewshotExampleSingleLabel), TypeError(
                    example_type_error_text.format(example_type=FewshotExampleSingleLabel, mode=self._mode)
                )
                if fs_example.label not in self._labels:
                    raise ValueError(
                        label_error_text.format(task_id=self.id, labels=self._labels, example_labels=(fs_example.label))
                    )

    @property
    def _state(self) -> dict[str, Any]:
        # Store labels as dict if descriptions exist, otherwise as list
        labels = self._label_descriptions if self._label_descriptions else self._labels
        return {
            **super()._state,
            "labels": labels,
        }

    @staticmethod
    def _result_to_scores(result: ResultMultiLabel | ResultSingleLabel) -> dict[str, float]:
        """Normalize a single result to a mapping of label → score.

        :params result: One result value from ``doc.results``.

        :return: Mapping from label to score.

        :raises TypeError: If the result has an unsupported type or shape.

        """
        if isinstance(result, ResultMultiLabel):
            return {str(label): float(score) for label, score in result.label_scores}

        if isinstance(result, ResultSingleLabel):
            return {result.label: result.score}

        raise TypeError(f"Unsupported result type in _result_to_scores: {type(result)}")

    @override
    def distill(
        self,
        base_model_id: str,
        framework: DistillationFramework,
        data: datasets.Dataset | Sequence[Doc],
        output_path: Path | str,
        val_frac: float,
        init_kwargs: dict[str, Any] | None = None,
        train_kwargs: dict[str, Any] | None = None,
        seed: int | None = None,
    ) -> None:
        init_kwargs = init_kwargs or {}
        train_kwargs = train_kwargs or {}
        output_path = Path(output_path)
        output_path.mkdir(parents=True, exist_ok=True)

        data = self.to_hf_dataset(data) if isinstance(data, Sequence) else data

        required_columns = {"text", "labels"}
        if not required_columns.issubset(data.column_names):
            raise ValueError(f"Dataset must contain columns: {required_columns}. Found: {data.column_names}")

        dataset_splits = self._split_dataset(data, 1 - val_frac, val_frac, seed)
        dataset_splits.save_to_disk(output_path / "data")

        match framework:
            case DistillationFramework.setfit:
                default_init_kwargs: dict[str, Any] = {}
                metric_kwargs: dict[str, Any] = {}

                if self._mode == "multi":
                    default_init_kwargs["multi_target_strategy"] = "multi-output"
                    metric_kwargs = {"average": "macro"}

                model = setfit.SetFitModel.from_pretrained(base_model_id, **(default_init_kwargs | init_kwargs))

                args = setfit.TrainingArguments(
                    output_dir=str(output_path),
                    eval_strategy="epoch",
                    save_strategy="epoch",
                    load_best_model_at_end=True,
                    **train_kwargs,
                )

                trainer = setfit.Trainer(
                    model=model,
                    args=args,
                    train_dataset=dataset_splits["train"],
                    eval_dataset=dataset_splits.get("val"),
                    metric="f1",
                    column_mapping={"text": "text", "labels": "label"},
                    metric_kwargs=metric_kwargs,
                )
                trainer.train()
                trainer.model.save_pretrained(output_path)

                metrics = trainer.evaluate()
                with open(output_path / "metrics.json", "w") as f:
                    json.dump(metrics, f, indent=4)

            case DistillationFramework.model2vec:

                def one_hot_to_label(label_indices: list[int]) -> list[str]:
                    """Convert list of label indices into list of labels.

                    :param label_indices: List of label indices.
                    :return: List of labels.
                    """
                    return [str(self._labels[i]) for i, is_label in enumerate(label_indices) if is_label]

                classifier = model2vec.train.StaticModelForClassification.from_pretrained(
                    model_name=base_model_id, **init_kwargs
                )
                classifier.fit(
                    dataset_splits["train"]["text"],
                    [one_hot_to_label(encoded_labels) for encoded_labels in dataset_splits["train"]["labels"]],
                    **train_kwargs,
                )
                classifier.to_pipeline().save_pretrained(output_path)

                metrics = classifier.evaluate(
                    dataset_splits["val"]["text"],
                    [one_hot_to_label(encoded_labels) for encoded_labels in dataset_splits["val"]["labels"]],
                )
                with open(output_path / "metrics.json", "w") as f:
                    json.dump(metrics, f, indent=4)

            case _:
                raise NotImplementedError(
                    f"Unsupported distillation framework for this task: {framework}. "
                    f"Please choose one of {DistillationFramework.setfit, DistillationFramework.model2vec}"
                )

    def to_hf_dataset(self, docs: Iterable[Doc], threshold: float | None = None) -> datasets.Dataset:
        """Convert results to a Hugging Face dataset with multi-hot labels.

        The emitted dataset contains a ``text`` column and a ``labels`` column which is a multi-hot list aligned to
        ``self._labels``. This method is robust to different result shapes produced by various model wrappers and
        bridges in both single-label and multi-label configurations:
        - ``list[tuple[str, float]]`` for multi-label results
        - ``tuple[str, float]`` for single-label results
        - ``str`` for single-label results (assumes score ``1.0``)
        - ``pydantic.BaseModel`` exposing ``label`` and optional ``score``

        :param docs: Documents whose ``results`` contain outputs for this task id.
        :param threshold: Threshold to convert scores into multi-hot indicators. Defaults to `THRESHOLD`.

        :return: A ``datasets.Dataset`` with ``text`` and multi-hot ``labels``.

        :raises KeyError: If any document is missing this task's results.
        :raises TypeError: If a result cannot be interpreted.

        """
        threshold = threshold or self.THRESHOLD

        data: list[dict[str, str | list[bool]]] = []

        # Define metadata and features (multi-hot across declared labels for multi-label).
        if self._mode == "multi":
            features = datasets.Features(
                {"text": datasets.Value("string"), "labels": datasets.Sequence(datasets.Value("bool"))}
            )
        else:
            features = datasets.Features(
                {"text": datasets.Value("string"), "labels": datasets.ClassLabel(names=self._labels)}
            )

        info = datasets.DatasetInfo(
            description=(
                f"{'Multi-label' if self._mode == 'multi' else 'Single-label'} classification dataset with labels "
                f"{self._labels}. Generated with sieves v{Config.get_version()}."
            ),
            features=features,
        )

        try:
            for doc in docs:
                scores = Classification._result_to_scores(doc.results[self._task_id])

                # If multi-label: store one-hot representation.
                if self._mode == "multi":
                    result_normalized = [int(scores.get(label, 0.0) >= threshold) for label in self._labels]  # type: ignore[no-matching-overload]
                # If single-label: get single-label result as is.
                else:
                    keys = list(scores.keys())
                    assert len(keys) == 1
                    result_normalized = keys[0]

                data.append({"text": doc.text, "labels": result_normalized})

        except KeyError as err:
            raise KeyError(f"Not all documents have results for this task with ID {self._task_id}") from err

        return datasets.Dataset.from_list(data, features=features, info=info)

    def _task_result_to_pydantic(self, result: Any) -> ResultSingleLabel | ResultMultiLabel:
        """Convert a result to the appropriate Pydantic model for this task.

        :param result: Input result.
        :return: Normalized Pydantic result.
        """
        if isinstance(result, ResultSingleLabel | ResultMultiLabel):
            return result

        # Use the same logic as `_task_result_to_dspy_dict` but return the Pydantic model.
        if isinstance(result, str):
            return ResultSingleLabel(label=result, score=1.0)
        elif isinstance(result, list) and all(isinstance(item, str) for item in result):
            if self._mode == "single":
                if len(result) == 1:
                    return ResultSingleLabel(label=result[0], score=1.0)
                else:
                    raise ValueError(f"Got list of {len(result)} labels for single-label task.")
            else:
                label_scores: list[tuple[str, float]] = []
                active_set = set(result)
                for label in self._labels:
                    s = 1.0 if label in active_set else 0.0
                    label_scores.append((label, s))
                return ResultMultiLabel(label_scores=label_scores)

        raise TypeError(f"Unsupported result type in classification task: {type(result)}")

    @override
    def _task_result_to_dspy_dict(self, result: Any) -> dict[str, Any]:
        # We accept `str` and `list[str]` for better UX.
        if isinstance(result, str):
            result = ResultSingleLabel(label=result, score=1.0)
        elif isinstance(result, list) and all(isinstance(item, str) for item in result):
            if self._mode == "single":
                if len(result) == 1:
                    result = ResultSingleLabel(label=result[0], score=1.0)
                else:
                    raise ValueError(f"Got list of {len(result)} labels for single-label task.")
            else:
                label_scores: list[tuple[str, float]] = []
                active_set = set(result)
                for label in self._labels:
                    s = 1.0 if label in active_set else 0.0
                    label_scores.append((label, s))
                result = ResultMultiLabel(label_scores=label_scores)

        assert isinstance(result, ResultSingleLabel | ResultMultiLabel)

        scores = self._result_to_scores(result)
        if self._mode == "multi":
            return {"score_per_label": scores}
        else:
            # For single label, it returns a dict with one key.
            label = list(scores.keys())[0]
            score = scores[label]
            return {"label": label, "score": score}

    @override
    def _evaluate_dspy_example(self, truth: dspy.Example, pred: dspy.Prediction, trace: Any, model: dspy.LM) -> float:
        if self._mode == "single":
            return 1.0 if truth["label"] == pred["label"] else 0.0

        # For multi-label: compute label-wise accuracy as
        # 1 - abs(true score for label - predicted score for label)
        # and normalize the sum of label-wise accuracies over all labels.
        accuracy = 0
        for label, score in truth["score_per_label"].items():
            if label in pred["score_per_label"]:
                pred_score = max(min(pred["score_per_label"][label], 1), 0)
                accuracy += 1 - abs(score - pred_score)

        return accuracy / len(truth["score_per_label"])

fewshot_example_type property

Return few-shot example type.

Returns:

Type	Description
type[FewshotExample]	Few-shot example type.

fewshot_examples property

Return few-shot examples.

Returns:

Type	Description
Sequence[FewshotExample]	Few-shot examples.

id property

Return task ID.

Used by pipeline for results and dependency management.

Returns:

Type	Description
str	Task ID.

prompt_signature_description property

Return prompt signature description.

Returns:

Type	Description
str | None	Prompt signature description.

prompt_template property

Return prompt template.

Returns:

Type	Description
str	Prompt template.

__add__(other)

Chain this task with another task or pipeline using the + operator.

This returns a new Pipeline that executes this task first, followed by the task(s) in other. The original task(s)/pipeline are not mutated.

Cache semantics: - If other is a Pipeline, the resulting pipeline adopts other's use_cache setting (because the left-hand side is a single task). - If other is a Task, the resulting pipeline defaults to use_cache=True.

Parameters:

Name	Type	Description	Default
other	Task | Pipeline	A Task or Pipeline to execute after this task.	required

Returns:

Type	Description
Pipeline	A new Pipeline representing the chained execution.

Raises:

Type	Description
TypeError	If other is not a Task or Pipeline.

Source code in sieves/tasks/core.py

def __add__(self, other: Task | Pipeline) -> Pipeline:
    """Chain this task with another task or pipeline using the ``+`` operator.

    This returns a new ``Pipeline`` that executes this task first, followed by the
    task(s) in ``other``. The original task(s)/pipeline are not mutated.

    Cache semantics:
    - If ``other`` is a ``Pipeline``, the resulting pipeline adopts ``other``'s
      ``use_cache`` setting (because the left-hand side is a single task).
    - If ``other`` is a ``Task``, the resulting pipeline defaults to ``use_cache=True``.

    :param other: A ``Task`` or ``Pipeline`` to execute after this task.
    :return: A new ``Pipeline`` representing the chained execution.
    :raises TypeError: If ``other`` is not a ``Task`` or ``Pipeline``.
    """
    # Lazy import to avoid circular dependency at module import time.
    from sieves.pipeline import Pipeline

    if isinstance(other, Pipeline):
        return Pipeline(tasks=[self, *other.tasks], use_cache=other.use_cache)

    if isinstance(other, Task):
        return Pipeline(tasks=[self, other])

    raise TypeError(f"Cannot chain Task with {type(other).__name__}")

__call__(docs)

Execute task with conditional logic.

Checks the condition for each document without materializing all docs upfront. Passes all documents that pass the condition to _call() for proper batching. Documents that fail the condition have results[task_id] set to None.

Parameters:

Name	Type	Description	Default
docs	Iterable[Doc]	Docs to process.	required

Returns:

Type	Description
Iterable[Doc]	Processed docs (in original order).

Source code in sieves/tasks/core.py

def __call__(self, docs: Iterable[Doc]) -> Iterable[Doc]:
    """Execute task with conditional logic.

    Checks the condition for each document without materializing all docs upfront.
    Passes all documents that pass the condition to _call() for proper batching.
    Documents that fail the condition have results[task_id] set to None.

    :param docs: Docs to process.
    :return: Processed docs (in original order).
    """
    docs = iter(docs) if not isinstance(docs, Iterator) else docs

    # Materialize docs in batches. This doesn't incur additional memory overhead, as docs are materialized in
    # batches downstream anyway.
    batch_size = self._batch_size if self._batch_size > 0 else sys.maxsize
    while docs_batch := [doc for doc in itertools.islice(docs, batch_size)]:
        # First pass: determine which docs pass the condition by index.
        passing_indices: set[int] = {
            idx for idx, doc in enumerate(docs_batch) if self._condition is None or self._condition(doc)
        }

        # Process all passing docs in one batch.
        processed = self._call(d for i, d in enumerate(docs_batch) if i in passing_indices)
        processed_iter = iter(processed) if not isinstance(processed, Iterator) else processed

        # Iterate through original docs in order and yield results.
        for idx, doc in enumerate(docs_batch):
            if idx in passing_indices:
                # Doc passed condition - use processed result.
                yield next(processed_iter)
            else:
                # Doc failed condition - set `None` result and yield original.
                doc.results[self.id] = None
                yield doc

__init__(labels, model, task_id=None, include_meta=True, batch_size=-1, prompt_instructions=None, fewshot_examples=(), mode='multi', model_settings=ModelSettings(), condition=None)

Initialize new Classification task.

Parameters:

Name	Type	Description	Default
labels	Sequence[str] | dict[str, str]	Labels to predict. Supports two formats: - List format: ["science", "politics", "sports"] - Dict format: {"science": "Scientific topics", "politics": "Political topics"} The dict format allows you to provide descriptions that help the model better understand each label.	required
model	TaskModel	Model to use.	required
task_id	str | None	Task ID.	None
include_meta	bool	Whether to include meta information generated by the task.	True
batch_size	int	Batch size to use for inference. Use -1 to process all documents at once.	-1
prompt_instructions	str | None	Custom prompt instructions. If None, default instructions are used.	None
fewshot_examples	Sequence[FewshotExample]	Few-shot examples.	()
mode	Literal['single', 'multi']	If 'multi', task returns confidence scores for all specified labels. If 'single', task returns most likely class label.	'multi'
model_settings	ModelSettings	Model settings.	ModelSettings()
condition	Callable[[Doc], bool] | None	Optional callable that determines whether to process each document.	None

Source code in sieves/tasks/predictive/classification/core.py

def __init__(
    self,
    labels: Sequence[str] | dict[str, str],
    model: TaskModel,
    task_id: str | None = None,
    include_meta: bool = True,
    batch_size: int = -1,
    prompt_instructions: str | None = None,
    fewshot_examples: Sequence[FewshotExample] = (),
    mode: Literal["single", "multi"] = "multi",
    model_settings: ModelSettings = ModelSettings(),
    condition: Callable[[Doc], bool] | None = None,
) -> None:
    """Initialize new Classification task.

    :param labels: Labels to predict. Supports two formats:
        - List format: `["science", "politics", "sports"]`
        - Dict format: `{"science": "Scientific topics", "politics": "Political topics"}`
        The dict format allows you to provide descriptions that help the model better understand each label.
    :param model: Model to use.
    :param task_id: Task ID.
    :param include_meta: Whether to include meta information generated by the task.
    :param batch_size: Batch size to use for inference. Use -1 to process all documents at once.
    :param prompt_instructions: Custom prompt instructions. If None, default instructions are used.
    :param fewshot_examples: Few-shot examples.
    :param mode: If 'multi', task returns confidence scores for all specified labels. If 'single', task returns
        most likely class label.
    :param model_settings: Model settings.
    :param condition: Optional callable that determines whether to process each document.
    """
    if isinstance(labels, dict):
        self._labels = list(labels.keys())
        self._label_descriptions = labels
    else:
        self._labels = list(labels)
        self._label_descriptions = {}
    self._mode = mode

    super().__init__(
        model=model,
        task_id=task_id,
        include_meta=include_meta,
        batch_size=batch_size,
        overwrite=False,
        prompt_instructions=prompt_instructions,
        fewshot_examples=fewshot_examples,
        model_settings=model_settings,
        condition=condition,
    )

deserialize(config, **kwargs) classmethod

Generate PredictiveTask instance from config.

Parameters:

Name	Type	Description	Default
config	Config	Config to generate instance from.	required
kwargs	dict[str, Any]	Values to inject into loaded config.	{}

Returns:

Type	Description
PredictiveTask[TaskPromptSignature, TaskResult, TaskBridge]	Deserialized PredictiveTask instance.

Source code in sieves/tasks/predictive/core.py

@classmethod
def deserialize(
    cls, config: Config, **kwargs: dict[str, Any]
) -> PredictiveTask[TaskPromptSignature, TaskResult, TaskBridge]:
    """Generate PredictiveTask instance from config.

    :param config: Config to generate instance from.
    :param kwargs: Values to inject into loaded config.
    :return PredictiveTask[TaskPromptSignature, TaskResult, _TaskBridge]: Deserialized PredictiveTask instance.
    """
    init_dict = config.to_init_dict(cls, **kwargs)
    init_dict["model_settings"] = ModelSettings.model_validate(init_dict["model_settings"])

    return cls(**init_dict)

evaluate(docs, judge=None, failure_threshold=0.5)

Evaluate task performance using DSPy-based evaluation.

Parameters:

Name	Type	Description	Default
docs	Iterable[Doc]	Documents to evaluate.	required
judge	LM | None	Optional DSPy LM instance to use as judge for generative tasks.	None
failure_threshold	float	Decision threshold for whether to mark predicitions as failures.	0.5

Returns:

Type	Description
TaskEvaluationReport	Evaluation report.

Source code in sieves/tasks/predictive/core.py

def evaluate(
    self, docs: Iterable[Doc], judge: dspy.LM | None = None, failure_threshold: float = 0.5
) -> TaskEvaluationReport:
    """Evaluate task performance using DSPy-based evaluation.

    :param docs: Documents to evaluate.
    :param judge: Optional DSPy LM instance to use as judge for generative tasks.
    :param failure_threshold: Decision threshold for whether to mark predicitions as failures.
    :return: Evaluation report.
    """
    truths: list[Any] = []
    preds: list[Any] = []
    failures: list[Doc] = []

    # Evaluate each doc individually to identify failed predictions.
    for doc in docs:
        if self.id not in doc.results:
            continue

        pred = doc.results[self.id]
        gold = doc.gold.get(self.id, None)

        # Accumulate for corpus-level metrics.
        truths.append(gold)
        preds.append(pred)

        # If gold or prediction is None: we cannot do proper evalution, so we just check whether they're both None
        # to compute score for failure analysis.
        if gold is None or pred is None:
            if gold is not None or pred is not None:
                failures.append(doc)
        else:
            # Convert result and gold to DSPy representation.
            truth = dspy.Example(**self._task_result_to_dspy_dict(gold))
            pred_dspy = dspy.Prediction(**self._task_result_to_dspy_dict(pred))

            # Call internal evaluation logic for per-doc failure analysis.
            score = self._evaluate_dspy_example(truth, pred_dspy, trace=None, model=judge)

            if score < failure_threshold:
                failures.append(doc)

    # Evaluate on corpus level to obtain representative metrics.
    metrics = self._compute_metrics(truths, preds, judge=judge)

    return TaskEvaluationReport(
        metrics=metrics,
        task_id=self.id,
        failures=failures,
    )

optimize(optimizer, verbose=True)

Optimize task prompt and few-shot examples with the available optimization config.

Updates task to use best prompt and few-shot examples found by the optimizer.

Parameters:

Name	Type	Description	Default
optimizer	Optimizer	Optimizer to run.	required
verbose	bool	Whether to suppress output. DSPy produces a good amount of logs, so this can be useful to not pollute your terminal. Only warnings and errors will be printed.	True

Returns:

Type	Description
tuple[str, Sequence[FewshotExample]]	Best found prompt and few-shot examples.

Source code in sieves/tasks/predictive/core.py

def optimize(self, optimizer: optimization.Optimizer, verbose: bool = True) -> tuple[str, Sequence[FewshotExample]]:
    """Optimize task prompt and few-shot examples with the available optimization config.

    Updates task to use best prompt and few-shot examples found by the optimizer.

    :param optimizer: Optimizer to run.
    :param verbose: Whether to suppress output. DSPy produces a good amount of logs, so this can be useful to
        not pollute your terminal. Only warnings and errors will be printed.

    :return tuple[str, Sequence[FewshotExample]]: Best found prompt and few-shot examples.
    """
    assert len(self._fewshot_examples) > 1, "At least two few-shot examples need to be provided to optimize."

    # Run optimizer to get best prompt and few-shot examples.
    signature = self._get_task_signature()
    dspy_examples = [ex.to_dspy() for ex in self._fewshot_examples]

    def _pred_eval(truth: dspy.Example, pred: dspy.Prediction, trace: Any | None = None) -> float:
        """Wrap optimization evaluation, inject model.

        :param truth: Ground truth.
        :param pred: Predicted value.
        :param trace: Optional trace information.
        :return: Metric value between 0.0 and 1.0.
        :raises KeyError: If target fields are missing from truth or prediction.
        :raises ValueError: If similarity score cannot be parsed from LLM response.
        """
        return self._evaluate_dspy_example(truth, pred, trace, model=optimizer.model)

    if verbose:
        best_prompt, best_examples = optimizer(signature, dspy_examples, _pred_eval, verbose=verbose)
    else:
        # Temporarily suppress DSPy logs.
        dspy_logger = logging.getLogger("dspy")
        optuna_logger = logging.getLogger("optuna")
        original_dspy_level = dspy_logger.level
        original_optuna_level = optuna_logger.level

        try:
            dspy_logger.setLevel(logging.ERROR)
            optuna_logger.setLevel(logging.ERROR)
            with warnings.catch_warnings():
                warnings.simplefilter("ignore")
                best_prompt, best_examples = optimizer(signature, dspy_examples, _pred_eval, verbose=verbose)
        finally:
            dspy_logger.setLevel(original_dspy_level)
            optuna_logger.setLevel(original_optuna_level)

    # Update few-shot examples and prompt instructions.
    fewshot_example_cls = self._fewshot_examples[0].__class__
    self._fewshot_examples = [fewshot_example_cls.from_dspy(ex) for ex in best_examples]
    self._validate_fewshot_examples()
    self._custom_prompt_instructions = best_prompt

    # Reinitialize bridge to use new prompt and few-shot examples.
    self._bridge = self._init_bridge(ModelType.get_model_type(self._model_wrapper))

    return best_prompt, self._fewshot_examples

serialize()

Serialize task.

Returns:

Type	Description
Config	Config instance.

Source code in sieves/tasks/core.py

def serialize(self) -> Config:
    """Serialize task.

    :return: Config instance.
    """
    return Config.create(self.__class__, {k: Attribute(value=v) for k, v in self._state.items()})

to_hf_dataset(docs, threshold=None)

Convert results to a Hugging Face dataset with multi-hot labels.

The emitted dataset contains a text column and a labels column which is a multi-hot list aligned to self._labels. This method is robust to different result shapes produced by various model wrappers and bridges in both single-label and multi-label configurations: - list[tuple[str, float]] for multi-label results - tuple[str, float] for single-label results - str for single-label results (assumes score 1.0) - pydantic.BaseModel exposing label and optional score

Parameters:

Name	Type	Description	Default
docs	Iterable[Doc]	Documents whose results contain outputs for this task id.	required
threshold	float | None	Threshold to convert scores into multi-hot indicators. Defaults to THRESHOLD.	None

Returns:

Type	Description
Dataset	A datasets.Dataset with text and multi-hot labels.

Raises:

Type	Description
KeyError	If any document is missing this task's results.
TypeError	If a result cannot be interpreted.

Source code in sieves/tasks/predictive/classification/core.py

def to_hf_dataset(self, docs: Iterable[Doc], threshold: float | None = None) -> datasets.Dataset:
    """Convert results to a Hugging Face dataset with multi-hot labels.

    The emitted dataset contains a ``text`` column and a ``labels`` column which is a multi-hot list aligned to
    ``self._labels``. This method is robust to different result shapes produced by various model wrappers and
    bridges in both single-label and multi-label configurations:
    - ``list[tuple[str, float]]`` for multi-label results
    - ``tuple[str, float]`` for single-label results
    - ``str`` for single-label results (assumes score ``1.0``)
    - ``pydantic.BaseModel`` exposing ``label`` and optional ``score``

    :param docs: Documents whose ``results`` contain outputs for this task id.
    :param threshold: Threshold to convert scores into multi-hot indicators. Defaults to `THRESHOLD`.

    :return: A ``datasets.Dataset`` with ``text`` and multi-hot ``labels``.

    :raises KeyError: If any document is missing this task's results.
    :raises TypeError: If a result cannot be interpreted.

    """
    threshold = threshold or self.THRESHOLD

    data: list[dict[str, str | list[bool]]] = []

    # Define metadata and features (multi-hot across declared labels for multi-label).
    if self._mode == "multi":
        features = datasets.Features(
            {"text": datasets.Value("string"), "labels": datasets.Sequence(datasets.Value("bool"))}
        )
    else:
        features = datasets.Features(
            {"text": datasets.Value("string"), "labels": datasets.ClassLabel(names=self._labels)}
        )

    info = datasets.DatasetInfo(
        description=(
            f"{'Multi-label' if self._mode == 'multi' else 'Single-label'} classification dataset with labels "
            f"{self._labels}. Generated with sieves v{Config.get_version()}."
        ),
        features=features,
    )

    try:
        for doc in docs:
            scores = Classification._result_to_scores(doc.results[self._task_id])

            # If multi-label: store one-hot representation.
            if self._mode == "multi":
                result_normalized = [int(scores.get(label, 0.0) >= threshold) for label in self._labels]  # type: ignore[no-matching-overload]
            # If single-label: get single-label result as is.
            else:
                keys = list(scores.keys())
                assert len(keys) == 1
                result_normalized = keys[0]

            data.append({"text": doc.text, "labels": result_normalized})

    except KeyError as err:
        raise KeyError(f"Not all documents have results for this task with ID {self._task_id}") from err

    return datasets.Dataset.from_list(data, features=features, info=info)

Bridges for classification task.

ClassificationBridge

Bases: Bridge[_BridgePromptSignature, _BridgeResult, ModelWrapperInferenceMode], ABC

Abstract base class for classification bridges.

Source code in sieves/tasks/predictive/classification/bridges.py

class ClassificationBridge(Bridge[_BridgePromptSignature, _BridgeResult, ModelWrapperInferenceMode], abc.ABC):
    """Abstract base class for classification bridges."""

    def __init__(
        self,
        task_id: str,
        prompt_instructions: str | None,
        labels: list[str] | dict[str, str],
        mode: Literal["single", "multi"],
        model_settings: ModelSettings,
        prompt_signature: type[pydantic.BaseModel],
        model_type: ModelType,
        fewshot_examples: Sequence[pydantic.BaseModel] = (),
    ):
        """Initialize ClassificationBridge.

        :param task_id: Task ID.
        :param prompt_instructions: Custom prompt instructions. If None, default instructions are used.
        :param labels: Labels to classify. Can be a list of label strings, or a dict mapping labels to descriptions.
        :param mode: If 'multi'', task returns scores for all specified labels. If 'single', task returns
        most likely class label.
        :param model_settings: Model settings.
        :param prompt_signature: Unified Pydantic prompt signature.
        :param model_type: Model type.
        :param fewshot_examples: Few-shot examples.
        """
        super().__init__(
            task_id=task_id,
            prompt_instructions=prompt_instructions,
            overwrite=False,
            model_settings=model_settings,
            prompt_signature=prompt_signature,
            model_type=model_type,
            fewshot_examples=fewshot_examples,
        )
        if isinstance(labels, dict):
            self._labels = list(labels.keys())
            self._label_descriptions = labels
        else:
            self._labels = labels
            self._label_descriptions = {}

        self._mode = mode
        self._consolidation_strategy = LabelScoreConsolidation(
            labels=self._labels,
            mode=self._mode,
            extractor=self._chunk_extractor,
        )

    @override
    @property
    def prompt_signature(self) -> _BridgePromptSignature:
        return convert_to_signature(
            model_cls=self._pydantic_signature,
            model_type=self.model_type,
            mode="classification",
        )  # type: ignore[return-value]

    @property
    @abc.abstractmethod
    def _chunk_extractor(self) -> Callable[[Any], dict[str, float]]:
        """Return a callable that extracts label scores from a raw chunk result.

        :return: Extractor callable.
        """

    def _get_label_descriptions(self) -> str:
        """Return a string with the label descriptions.

        :return: A string with the label descriptions.
        """
        labels_with_descriptions: list[str] = []
        for label in self._labels:
            if label in self._label_descriptions:
                labels_with_descriptions.append(
                    f"  <label_description>\n    <label>{label}</label>\n    <description>"
                    f"{self._label_descriptions[label]}</description>\n  </label_description>"
                )
            else:
                labels_with_descriptions.append(f"  <label>{label}</label>")

        label_desc_string = "\n".join(labels_with_descriptions)

        return f"<label_descriptions>\n{label_desc_string}\n</label_descriptions>"

inference_mode abstractmethod property

Return inference mode.

Returns:

Type	Description
ModelWrapperInferenceMode	Inference mode.

model_settings property

Return model settings.

Returns:

Type	Description
ModelSettings	Model settings.

model_type property

Return model type.

Returns:

Type	Description
ModelType	Model type.

prompt_template property

Return prompt template.

Chains _prompt_instructions, _prompt_example_xml and _prompt_conclusion.

Note: different model have different expectations as to how a prompt should look like. E.g. outlines supports the Jinja 2 templating format for insertion of values and few-shot examples, whereas DSPy integrates these things in a different value in the workflow and hence expects the prompt not to include these things. Mind model-specific expectations when creating a prompt template.

Returns:

Type	Description
str	Prompt template as string. None if not used by model wrapper.

__init__(task_id, prompt_instructions, labels, mode, model_settings, prompt_signature, model_type, fewshot_examples=())

Initialize ClassificationBridge.

Parameters:

Name	Type	Description	Default
task_id	str	Task ID.	required
prompt_instructions	str | None	Custom prompt instructions. If None, default instructions are used.	required
labels	list[str] | dict[str, str]	Labels to classify. Can be a list of label strings, or a dict mapping labels to descriptions.	required
mode	Literal['single', 'multi']	If 'multi'', task returns scores for all specified labels. If 'single', task returns most likely class label.	required
model_settings	ModelSettings	Model settings.	required
prompt_signature	type[BaseModel]	Unified Pydantic prompt signature.	required
model_type	ModelType	Model type.	required
fewshot_examples	Sequence[BaseModel]	Few-shot examples.	()

Source code in sieves/tasks/predictive/classification/bridges.py

def __init__(
    self,
    task_id: str,
    prompt_instructions: str | None,
    labels: list[str] | dict[str, str],
    mode: Literal["single", "multi"],
    model_settings: ModelSettings,
    prompt_signature: type[pydantic.BaseModel],
    model_type: ModelType,
    fewshot_examples: Sequence[pydantic.BaseModel] = (),
):
    """Initialize ClassificationBridge.

    :param task_id: Task ID.
    :param prompt_instructions: Custom prompt instructions. If None, default instructions are used.
    :param labels: Labels to classify. Can be a list of label strings, or a dict mapping labels to descriptions.
    :param mode: If 'multi'', task returns scores for all specified labels. If 'single', task returns
    most likely class label.
    :param model_settings: Model settings.
    :param prompt_signature: Unified Pydantic prompt signature.
    :param model_type: Model type.
    :param fewshot_examples: Few-shot examples.
    """
    super().__init__(
        task_id=task_id,
        prompt_instructions=prompt_instructions,
        overwrite=False,
        model_settings=model_settings,
        prompt_signature=prompt_signature,
        model_type=model_type,
        fewshot_examples=fewshot_examples,
    )
    if isinstance(labels, dict):
        self._labels = list(labels.keys())
        self._label_descriptions = labels
    else:
        self._labels = labels
        self._label_descriptions = {}

    self._mode = mode
    self._consolidation_strategy = LabelScoreConsolidation(
        labels=self._labels,
        mode=self._mode,
        extractor=self._chunk_extractor,
    )

consolidate(results, docs_offsets) abstractmethod

Consolidate results for document chunks into document results.

Parameters:

Name	Type	Description	Default
results	Sequence[TaskResult]	Results per document chunk.	required
docs_offsets	list[tuple[int, int]]	Chunk offsets per document. Chunks per document can be obtained with results[docs_chunk_offsets[i][0]:docs_chunk_offsets[i][1]].	required

Returns:

Type	Description
Sequence[TaskResult]	Results per document as a sequence.

Source code in sieves/tasks/predictive/bridges.py

@abc.abstractmethod
def consolidate(self, results: Sequence[TaskResult], docs_offsets: list[tuple[int, int]]) -> Sequence[TaskResult]:
    """Consolidate results for document chunks into document results.

    :param results: Results per document chunk.
    :param docs_offsets: Chunk offsets per document. Chunks per document can be obtained with
        `results[docs_chunk_offsets[i][0]:docs_chunk_offsets[i][1]]`.
    :return: Results per document as a sequence.
    """

extract(docs)

Extract all values from doc instances that are to be injected into the prompts.

Parameters:

Name	Type	Description	Default
docs	Sequence[Doc]	Docs to extract values from.	required

Returns:

Type	Description
Sequence[dict[str, Any]]	All values from doc instances that are to be injected into the prompts as a sequence.

Source code in sieves/tasks/predictive/bridges.py

def extract(self, docs: Sequence[Doc]) -> Sequence[dict[str, Any]]:
    """Extract all values from doc instances that are to be injected into the prompts.

    :param docs: Docs to extract values from.
    :return: All values from doc instances that are to be injected into the prompts as a sequence.
    """
    return [{"text": doc.text if doc.text else None} for doc in docs]

integrate(results, docs) abstractmethod

Integrate results into Doc instances.

Parameters:

Name	Type	Description	Default
results	Sequence[TaskResult]	Results from prompt executable.	required
docs	list[Doc]	Doc instances to update.	required

Returns:

Type	Description
list[Doc]	Updated doc instances as a list.

Source code in sieves/tasks/predictive/bridges.py

@abc.abstractmethod
def integrate(self, results: Sequence[TaskResult], docs: list[Doc]) -> list[Doc]:
    """Integrate results into Doc instances.

    :param results: Results from prompt executable.
    :param docs: Doc instances to update.
    :return: Updated doc instances as a list.
    """

DSPyClassification

Bases: ClassificationBridge[PromptSignature, Result, InferenceMode]

DSPy bridge for classification.

Source code in sieves/tasks/predictive/classification/bridges.py

class DSPyClassification(ClassificationBridge[dspy_.PromptSignature, dspy_.Result, dspy_.InferenceMode]):
    """DSPy bridge for classification."""

    @override
    def _validate(self) -> None:
        assert self._model_type == ModelType.dspy

    @override
    @property
    def _default_prompt_instructions(self) -> str:
        return ""

    @override
    @property
    def inference_mode(self) -> dspy_.InferenceMode:
        return self._model_settings.inference_mode or dspy_.InferenceMode.predict

    @property
    @override
    def _chunk_extractor(self) -> Callable[[Any], dict[str, float]]:
        def extractor(res: Any) -> dict[str, float]:
            if self._mode == "multi":
                return {label: getattr(res, label) for label in self._labels}
            return {res.label: res.score}

        return extractor

    @override
    def integrate(self, results: Sequence[dspy_.Result], docs: list[Doc]) -> list[Doc]:
        for doc, result in zip(docs, results):
            # The result is a dspy.Prediction where completions contain the fields.
            # We take the first completion.
            prediction = result.completions[0]

            if self._mode == "multi":
                label_scores = [(label, float(getattr(prediction, label))) for label in self._labels]
                sorted_preds = sorted(label_scores, key=lambda x: x[1], reverse=True)
                doc.results[self._task_id] = ResultMultiLabel(label_scores=sorted_preds)
            else:
                doc.results[self._task_id] = ResultSingleLabel(label=prediction.label, score=float(prediction.score))

        return docs

    @override
    def consolidate(
        self, results: Sequence[dspy_.Result], docs_offsets: list[tuple[int, int]]
    ) -> Sequence[dspy_.Result]:
        consolidated_results_clean = self._consolidation_strategy.consolidate(results, docs_offsets)

        # Wrap back into dspy.Prediction.
        consolidated_results: list[dspy_.Result] = []
        for scores_list in consolidated_results_clean:
            if self._mode == "multi":
                data = {label: score for label, score in scores_list}
            else:
                data = {"label": scores_list[0][0], "score": scores_list[0][1]}

            consolidated_results.append(
                dspy.Prediction.from_completions(
                    [data],
                    signature=self.prompt_signature,
                )
            )
        return consolidated_results

model_settings property

Return model settings.

Returns:

Type	Description
ModelSettings	Model settings.

model_type property

Return model type.

Returns:

Type	Description
ModelType	Model type.

prompt_template property

Return prompt template.

Chains _prompt_instructions, _prompt_example_xml and _prompt_conclusion.

Note: different model have different expectations as to how a prompt should look like. E.g. outlines supports the Jinja 2 templating format for insertion of values and few-shot examples, whereas DSPy integrates these things in a different value in the workflow and hence expects the prompt not to include these things. Mind model-specific expectations when creating a prompt template.

Returns:

Type	Description
str	Prompt template as string. None if not used by model wrapper.

__init__(task_id, prompt_instructions, labels, mode, model_settings, prompt_signature, model_type, fewshot_examples=())

Initialize ClassificationBridge.

Parameters:

Name	Type	Description	Default
task_id	str	Task ID.	required
prompt_instructions	str | None	Custom prompt instructions. If None, default instructions are used.	required
labels	list[str] | dict[str, str]	Labels to classify. Can be a list of label strings, or a dict mapping labels to descriptions.	required
mode	Literal['single', 'multi']	If 'multi'', task returns scores for all specified labels. If 'single', task returns most likely class label.	required
model_settings	ModelSettings	Model settings.	required
prompt_signature	type[BaseModel]	Unified Pydantic prompt signature.	required
model_type	ModelType	Model type.	required
fewshot_examples	Sequence[BaseModel]	Few-shot examples.	()

Source code in sieves/tasks/predictive/classification/bridges.py

def __init__(
    self,
    task_id: str,
    prompt_instructions: str | None,
    labels: list[str] | dict[str, str],
    mode: Literal["single", "multi"],
    model_settings: ModelSettings,
    prompt_signature: type[pydantic.BaseModel],
    model_type: ModelType,
    fewshot_examples: Sequence[pydantic.BaseModel] = (),
):
    """Initialize ClassificationBridge.

    :param task_id: Task ID.
    :param prompt_instructions: Custom prompt instructions. If None, default instructions are used.
    :param labels: Labels to classify. Can be a list of label strings, or a dict mapping labels to descriptions.
    :param mode: If 'multi'', task returns scores for all specified labels. If 'single', task returns
    most likely class label.
    :param model_settings: Model settings.
    :param prompt_signature: Unified Pydantic prompt signature.
    :param model_type: Model type.
    :param fewshot_examples: Few-shot examples.
    """
    super().__init__(
        task_id=task_id,
        prompt_instructions=prompt_instructions,
        overwrite=False,
        model_settings=model_settings,
        prompt_signature=prompt_signature,
        model_type=model_type,
        fewshot_examples=fewshot_examples,
    )
    if isinstance(labels, dict):
        self._labels = list(labels.keys())
        self._label_descriptions = labels
    else:
        self._labels = labels
        self._label_descriptions = {}

    self._mode = mode
    self._consolidation_strategy = LabelScoreConsolidation(
        labels=self._labels,
        mode=self._mode,
        extractor=self._chunk_extractor,
    )

extract(docs)

Extract all values from doc instances that are to be injected into the prompts.

Parameters:

Name	Type	Description	Default
docs	Sequence[Doc]	Docs to extract values from.	required

Returns:

Type	Description
Sequence[dict[str, Any]]	All values from doc instances that are to be injected into the prompts as a sequence.

Source code in sieves/tasks/predictive/bridges.py

def extract(self, docs: Sequence[Doc]) -> Sequence[dict[str, Any]]:
    """Extract all values from doc instances that are to be injected into the prompts.

    :param docs: Docs to extract values from.
    :return: All values from doc instances that are to be injected into the prompts as a sequence.
    """
    return [{"text": doc.text if doc.text else None} for doc in docs]

HuggingFaceClassification

Bases: ClassificationBridge[list[str], Result, InferenceMode]

HuggingFace bridge for classification.

Source code in sieves/tasks/predictive/classification/bridges.py

class HuggingFaceClassification(ClassificationBridge[list[str], huggingface_.Result, huggingface_.InferenceMode]):
    """HuggingFace bridge for classification."""

    @override
    def _validate(self) -> None:
        assert self._model_type == ModelType.huggingface

    @override
    @property
    def _default_prompt_instructions(self) -> str:
        return f"This text is about {{}}.\n{self._get_label_descriptions()}"

    @override
    @property
    def inference_mode(self) -> huggingface_.InferenceMode:
        return self._model_settings.inference_mode or huggingface_.InferenceMode.zeroshot_cls

    @property
    @override
    def _chunk_extractor(self) -> Callable[[Any], dict[str, float]]:
        # For HuggingFace zero-shot, prompt_signature is a list of field names.
        # convert_to_signature for HF returns all fields but 'score'.
        # The raw result 'res' from HF has 'labels' and 'scores'.
        return lambda res: dict(zip(res["labels"], res["scores"]))

    @override
    def integrate(self, results: Sequence[huggingface_.Result], docs: list[Doc]) -> list[Doc]:
        for doc, result in zip(docs, results):
            # result is a dict with 'labels' and 'scores'.
            # We map them back to the unified schema.
            # In multi-label mode, doc.results[self._task_id] expects a ResultMultiLabel (list of tuples).
            label_scores = list(zip(result["labels"], result["scores"]))
            sorted_preds = sorted(label_scores, key=lambda x: x[1], reverse=True)

            if self._mode == "multi":
                doc.results[self._task_id] = ResultMultiLabel(label_scores=sorted_preds)
            else:
                # In single mode, the top label should be in the ResultSingleLabel.
                # Usually HF zero-shot with mode='multi' (default in sieves) returns multiple labels.
                # ClassificationTask._init_bridge should ideally handle this.
                doc.results[self._task_id] = ResultSingleLabel(label=sorted_preds[0][0], score=sorted_preds[0][1])
        return docs

    @override
    def consolidate(
        self, results: Sequence[huggingface_.Result], docs_offsets: list[tuple[int, int]]
    ) -> Sequence[huggingface_.Result]:
        consolidated_results_clean = self._consolidation_strategy.consolidate(results, docs_offsets)

        consolidated_results: list[huggingface_.Result] = []
        for scores_list in consolidated_results_clean:
            consolidated_results.append(
                {
                    "labels": [label for label, _ in scores_list],
                    "scores": [score for _, score in scores_list],
                }
            )
        return consolidated_results

model_settings property

Return model settings.

Returns:

Type	Description
ModelSettings	Model settings.

model_type property

Return model type.

Returns:

Type	Description
ModelType	Model type.

prompt_template property

Return prompt template.

Chains _prompt_instructions, _prompt_example_xml and _prompt_conclusion.

Note: different model have different expectations as to how a prompt should look like. E.g. outlines supports the Jinja 2 templating format for insertion of values and few-shot examples, whereas DSPy integrates these things in a different value in the workflow and hence expects the prompt not to include these things. Mind model-specific expectations when creating a prompt template.

Returns:

Type	Description
str	Prompt template as string. None if not used by model wrapper.

__init__(task_id, prompt_instructions, labels, mode, model_settings, prompt_signature, model_type, fewshot_examples=())

Initialize ClassificationBridge.

Parameters:

Name	Type	Description	Default
task_id	str	Task ID.	required
prompt_instructions	str | None	Custom prompt instructions. If None, default instructions are used.	required
labels	list[str] | dict[str, str]	Labels to classify. Can be a list of label strings, or a dict mapping labels to descriptions.	required
mode	Literal['single', 'multi']	If 'multi'', task returns scores for all specified labels. If 'single', task returns most likely class label.	required
model_settings	ModelSettings	Model settings.	required
prompt_signature	type[BaseModel]	Unified Pydantic prompt signature.	required
model_type	ModelType	Model type.	required
fewshot_examples	Sequence[BaseModel]	Few-shot examples.	()

Source code in sieves/tasks/predictive/classification/bridges.py

def __init__(
    self,
    task_id: str,
    prompt_instructions: str | None,
    labels: list[str] | dict[str, str],
    mode: Literal["single", "multi"],
    model_settings: ModelSettings,
    prompt_signature: type[pydantic.BaseModel],
    model_type: ModelType,
    fewshot_examples: Sequence[pydantic.BaseModel] = (),
):
    """Initialize ClassificationBridge.

    :param task_id: Task ID.
    :param prompt_instructions: Custom prompt instructions. If None, default instructions are used.
    :param labels: Labels to classify. Can be a list of label strings, or a dict mapping labels to descriptions.
    :param mode: If 'multi'', task returns scores for all specified labels. If 'single', task returns
    most likely class label.
    :param model_settings: Model settings.
    :param prompt_signature: Unified Pydantic prompt signature.
    :param model_type: Model type.
    :param fewshot_examples: Few-shot examples.
    """
    super().__init__(
        task_id=task_id,
        prompt_instructions=prompt_instructions,
        overwrite=False,
        model_settings=model_settings,
        prompt_signature=prompt_signature,
        model_type=model_type,
        fewshot_examples=fewshot_examples,
    )
    if isinstance(labels, dict):
        self._labels = list(labels.keys())
        self._label_descriptions = labels
    else:
        self._labels = labels
        self._label_descriptions = {}

    self._mode = mode
    self._consolidation_strategy = LabelScoreConsolidation(
        labels=self._labels,
        mode=self._mode,
        extractor=self._chunk_extractor,
    )

extract(docs)

Extract all values from doc instances that are to be injected into the prompts.

Parameters:

Name	Type	Description	Default
docs	Sequence[Doc]	Docs to extract values from.	required

Returns:

Type	Description
Sequence[dict[str, Any]]	All values from doc instances that are to be injected into the prompts as a sequence.

Source code in sieves/tasks/predictive/bridges.py

def extract(self, docs: Sequence[Doc]) -> Sequence[dict[str, Any]]:
    """Extract all values from doc instances that are to be injected into the prompts.

    :param docs: Docs to extract values from.
    :return: All values from doc instances that are to be injected into the prompts as a sequence.
    """
    return [{"text": doc.text if doc.text else None} for doc in docs]

LangChainClassification

Bases: ClassificationBridge[BaseModel | list[str], BaseModel | str, ModelWrapperInferenceMode], ABC

Base class for Pydantic-based classification bridges.

Source code in sieves/tasks/predictive/classification/bridges.py

class LangChainClassification(
    ClassificationBridge[pydantic.BaseModel | list[str], pydantic.BaseModel | str, ModelWrapperInferenceMode], abc.ABC
):
    """Base class for Pydantic-based classification bridges."""

    @override
    def _validate(self) -> None:
        assert self._model_type == ModelType.langchain

    @override
    @property
    def _default_prompt_instructions(self) -> str:
        if self._mode == "multi":
            return """
            Perform multi-label classification of the provided text.
            For each label, provide a score between 0.0 (not applicable) and 1.0 (highly applicable).
            """

        return """
        Classify the provided text.
        Provide a score reflecting how likely it is that your chosen label is the correct
        fit for the text.
        """

    @override
    @property
    def _prompt_conclusion(self) -> str | None:
        return """
        ========

        <text>{{ text }}</text>
        """

    @property
    @override
    def _chunk_extractor(self) -> Callable[[Any], dict[str, float]]:
        def extractor(res: Any) -> dict[str, float]:
            if self._mode == "multi":
                return {label: float(getattr(res, label)) for label in self._labels}
            return {str(getattr(res, "label")): float(getattr(res, "score"))}

        return extractor

    @override
    def integrate(self, results: Sequence[pydantic.BaseModel | str], docs: list[Doc]) -> list[Doc]:
        for doc, result in zip(docs, results):
            if self._mode == "multi":
                assert isinstance(result, pydantic.BaseModel)
                label_scores = result.model_dump()
                sorted_label_scores = sorted(
                    ((label, score) for label, score in label_scores.items()), key=lambda x: x[1], reverse=True
                )
                doc.results[self._task_id] = ResultMultiLabel(label_scores=sorted_label_scores)

            else:
                assert hasattr(result, "label") and hasattr(result, "score")
                doc.results[self._task_id] = ResultSingleLabel(label=result.label, score=result.score)

        return docs

    @override
    def consolidate(
        self, results: Sequence[pydantic.BaseModel | str], docs_offsets: list[tuple[int, int]]
    ) -> Sequence[pydantic.BaseModel | str]:
        consolidated_results_clean = self._consolidation_strategy.consolidate(results, docs_offsets)

        consolidated_results: list[pydantic.BaseModel | str] = []
        prompt_signature = self.prompt_signature
        assert issubclass(prompt_signature, pydantic.BaseModel)  # type: ignore[arg-type]

        for scores_list in consolidated_results_clean:
            if self._mode == "multi":
                consolidated_results.append(prompt_signature(**dict(scores_list)))
            else:
                # In single mode, we only take the top label.
                top_label, top_score = scores_list[0]
                consolidated_results.append(
                    prompt_signature(
                        label=top_label,
                        score=top_score,
                    )
                )

        return consolidated_results

    @override
    @property
    def inference_mode(self) -> langchain_.InferenceMode:
        return self._model_settings.inference_mode or langchain_.InferenceMode.structured

model_settings property

Return model settings.

Returns:

Type	Description
ModelSettings	Model settings.

model_type property

Return model type.

Returns:

Type	Description
ModelType	Model type.

prompt_template property

Return prompt template.

Chains _prompt_instructions, _prompt_example_xml and _prompt_conclusion.

Note: different model have different expectations as to how a prompt should look like. E.g. outlines supports the Jinja 2 templating format for insertion of values and few-shot examples, whereas DSPy integrates these things in a different value in the workflow and hence expects the prompt not to include these things. Mind model-specific expectations when creating a prompt template.

Returns:

Type	Description
str	Prompt template as string. None if not used by model wrapper.

__init__(task_id, prompt_instructions, labels, mode, model_settings, prompt_signature, model_type, fewshot_examples=())

Initialize ClassificationBridge.

Parameters:

Name	Type	Description	Default
task_id	str	Task ID.	required
prompt_instructions	str | None	Custom prompt instructions. If None, default instructions are used.	required
labels	list[str] | dict[str, str]	Labels to classify. Can be a list of label strings, or a dict mapping labels to descriptions.	required
mode	Literal['single', 'multi']	If 'multi'', task returns scores for all specified labels. If 'single', task returns most likely class label.	required
model_settings	ModelSettings	Model settings.	required
prompt_signature	type[BaseModel]	Unified Pydantic prompt signature.	required
model_type	ModelType	Model type.	required
fewshot_examples	Sequence[BaseModel]	Few-shot examples.	()

Source code in sieves/tasks/predictive/classification/bridges.py

def __init__(
    self,
    task_id: str,
    prompt_instructions: str | None,
    labels: list[str] | dict[str, str],
    mode: Literal["single", "multi"],
    model_settings: ModelSettings,
    prompt_signature: type[pydantic.BaseModel],
    model_type: ModelType,
    fewshot_examples: Sequence[pydantic.BaseModel] = (),
):
    """Initialize ClassificationBridge.

    :param task_id: Task ID.
    :param prompt_instructions: Custom prompt instructions. If None, default instructions are used.
    :param labels: Labels to classify. Can be a list of label strings, or a dict mapping labels to descriptions.
    :param mode: If 'multi'', task returns scores for all specified labels. If 'single', task returns
    most likely class label.
    :param model_settings: Model settings.
    :param prompt_signature: Unified Pydantic prompt signature.
    :param model_type: Model type.
    :param fewshot_examples: Few-shot examples.
    """
    super().__init__(
        task_id=task_id,
        prompt_instructions=prompt_instructions,
        overwrite=False,
        model_settings=model_settings,
        prompt_signature=prompt_signature,
        model_type=model_type,
        fewshot_examples=fewshot_examples,
    )
    if isinstance(labels, dict):
        self._labels = list(labels.keys())
        self._label_descriptions = labels
    else:
        self._labels = labels
        self._label_descriptions = {}

    self._mode = mode
    self._consolidation_strategy = LabelScoreConsolidation(
        labels=self._labels,
        mode=self._mode,
        extractor=self._chunk_extractor,
    )

extract(docs)

Extract all values from doc instances that are to be injected into the prompts.

Parameters:

Name	Type	Description	Default
docs	Sequence[Doc]	Docs to extract values from.	required

Returns:

Type	Description
Sequence[dict[str, Any]]	All values from doc instances that are to be injected into the prompts as a sequence.

Source code in sieves/tasks/predictive/bridges.py

def extract(self, docs: Sequence[Doc]) -> Sequence[dict[str, Any]]:
    """Extract all values from doc instances that are to be injected into the prompts.

    :param docs: Docs to extract values from.
    :return: All values from doc instances that are to be injected into the prompts as a sequence.
    """
    return [{"text": doc.text if doc.text else None} for doc in docs]

OutlinesClassification

Bases: LangChainClassification[ModelWrapperInferenceMode], ABC

Base class for Outlines-based classification bridges with label forcing.

Source code in sieves/tasks/predictive/classification/bridges.py

class OutlinesClassification(LangChainClassification[ModelWrapperInferenceMode], abc.ABC):
    """Base class for Outlines-based classification bridges with label forcing."""

    @override
    def _validate(self) -> None:
        assert self._model_type == ModelType.outlines

    @override
    @property
    def _default_prompt_instructions(self) -> str:
        if self._mode == "multi":
            return super()._default_prompt_instructions

        return f"""
        Perform single-label classification of the provided text given the provided labels: {",".join(self._labels)}.
        {self._get_label_descriptions()}

        Provide the best-fitting label for given text.
        """

    @property
    @override
    def _chunk_extractor(self) -> Callable[[Any], dict[str, float]]:
        if self._mode == "multi":
            return super()._chunk_extractor

        def extractor(res: Any) -> dict[str, float]:
            if isinstance(res, str):
                return {res: 1.0}
            return {str(getattr(res, "label")): float(getattr(res, "score", 1.0))}

        return extractor

    @override
    def integrate(self, results: Sequence[pydantic.BaseModel | str], docs: list[Doc]) -> list[Doc]:
        if self._mode == "multi":
            return super().integrate(results, docs)

        for doc, result in zip(docs, results):
            # Outlines choice mode returns just the label string.
            if isinstance(result, str):
                doc.results[self._task_id] = ResultSingleLabel(label=result, score=1.0)
            else:
                # Fallback for other pydantic-based bridges.
                assert hasattr(result, "label")
                doc.results[self._task_id] = ResultSingleLabel(label=result.label, score=getattr(result, "score", 1.0))
        return docs

    @override
    def consolidate(
        self, results: Sequence[pydantic.BaseModel | str], docs_offsets: list[tuple[int, int]]
    ) -> Sequence[pydantic.BaseModel | str]:
        if self._mode == "multi":
            return super().consolidate(results, docs_offsets)

        consolidated_results_clean = self._consolidation_strategy.consolidate(results, docs_offsets)

        consolidated_results: list[pydantic.BaseModel | str] = []
        for scores_list in consolidated_results_clean:
            top_label, _ = scores_list[0]
            consolidated_results.append(top_label)

        return consolidated_results

    @override
    @property
    def inference_mode(self) -> outlines_.InferenceMode:
        return self._model_settings.inference_mode or (
            outlines_.InferenceMode.json if self._mode == "multi" else outlines_.InferenceMode.choice
        )

model_settings property

Return model settings.

Returns:

Type	Description
ModelSettings	Model settings.

model_type property

Return model type.

Returns:

Type	Description
ModelType	Model type.

prompt_template property

Return prompt template.

Chains _prompt_instructions, _prompt_example_xml and _prompt_conclusion.

Note: different model have different expectations as to how a prompt should look like. E.g. outlines supports the Jinja 2 templating format for insertion of values and few-shot examples, whereas DSPy integrates these things in a different value in the workflow and hence expects the prompt not to include these things. Mind model-specific expectations when creating a prompt template.

Returns:

Type	Description
str	Prompt template as string. None if not used by model wrapper.

__init__(task_id, prompt_instructions, labels, mode, model_settings, prompt_signature, model_type, fewshot_examples=())

Initialize ClassificationBridge.

Parameters:

Name	Type	Description	Default
task_id	str	Task ID.	required
prompt_instructions	str | None	Custom prompt instructions. If None, default instructions are used.	required
labels	list[str] | dict[str, str]	Labels to classify. Can be a list of label strings, or a dict mapping labels to descriptions.	required
mode	Literal['single', 'multi']	If 'multi'', task returns scores for all specified labels. If 'single', task returns most likely class label.	required
model_settings	ModelSettings	Model settings.	required
prompt_signature	type[BaseModel]	Unified Pydantic prompt signature.	required
model_type	ModelType	Model type.	required
fewshot_examples	Sequence[BaseModel]	Few-shot examples.	()

Source code in sieves/tasks/predictive/classification/bridges.py

def __init__(
    self,
    task_id: str,
    prompt_instructions: str | None,
    labels: list[str] | dict[str, str],
    mode: Literal["single", "multi"],
    model_settings: ModelSettings,
    prompt_signature: type[pydantic.BaseModel],
    model_type: ModelType,
    fewshot_examples: Sequence[pydantic.BaseModel] = (),
):
    """Initialize ClassificationBridge.

    :param task_id: Task ID.
    :param prompt_instructions: Custom prompt instructions. If None, default instructions are used.
    :param labels: Labels to classify. Can be a list of label strings, or a dict mapping labels to descriptions.
    :param mode: If 'multi'', task returns scores for all specified labels. If 'single', task returns
    most likely class label.
    :param model_settings: Model settings.
    :param prompt_signature: Unified Pydantic prompt signature.
    :param model_type: Model type.
    :param fewshot_examples: Few-shot examples.
    """
    super().__init__(
        task_id=task_id,
        prompt_instructions=prompt_instructions,
        overwrite=False,
        model_settings=model_settings,
        prompt_signature=prompt_signature,
        model_type=model_type,
        fewshot_examples=fewshot_examples,
    )
    if isinstance(labels, dict):
        self._labels = list(labels.keys())
        self._label_descriptions = labels
    else:
        self._labels = labels
        self._label_descriptions = {}

    self._mode = mode
    self._consolidation_strategy = LabelScoreConsolidation(
        labels=self._labels,
        mode=self._mode,
        extractor=self._chunk_extractor,
    )

extract(docs)

Extract all values from doc instances that are to be injected into the prompts.

Parameters:

Name	Type	Description	Default
docs	Sequence[Doc]	Docs to extract values from.	required

Returns:

Type	Description
Sequence[dict[str, Any]]	All values from doc instances that are to be injected into the prompts as a sequence.

Source code in sieves/tasks/predictive/bridges.py

def extract(self, docs: Sequence[Doc]) -> Sequence[dict[str, Any]]:
    """Extract all values from doc instances that are to be injected into the prompts.

    :param docs: Docs to extract values from.
    :return: All values from doc instances that are to be injected into the prompts as a sequence.
    """
    return [{"text": doc.text if doc.text else None} for doc in docs]

Schemas for classification task.

FewshotExampleMultiLabel

Bases: FewshotExample

Few‑shot example for multi‑label classification with per‑label scores.

Attributes: text: Input text. score_per_label: Mapping of labels to confidence scores.

Source code in sieves/tasks/predictive/schemas/classification.py

class FewshotExampleMultiLabel(BaseFewshotExample):
    """Few‑shot example for multi‑label classification with per‑label scores.

    Attributes:
        text: Input text.
        score_per_label: Mapping of labels to confidence scores.
    """

    score_per_label: dict[str, float]

    @pydantic.model_validator(mode="after")
    def check_score(self) -> FewshotExampleMultiLabel:
        """Validate that scores lie within [0, 1].

        :return: Validated instance.
        """
        if any([conf for conf in self.score_per_label.values() if not 0 <= conf <= 1]):
            raise ValueError("Score has to be between 0 and 1.")
        return self

    @property
    def target_fields(self) -> tuple[str, ...]:
        """Return target fields.

        :return: Target fields.
        """
        return ("score_per_label",)

input_fields property

Defines which fields are inputs.

Returns:

Type	Description
Sequence[str]	Sequence of field names.

target_fields property

Return target fields.

Returns:

Type	Description
tuple[str, ...]	Target fields.

check_score()

Validate that scores lie within [0, 1].

Returns:

Type	Description
FewshotExampleMultiLabel	Validated instance.

Source code in sieves/tasks/predictive/schemas/classification.py

@pydantic.model_validator(mode="after")
def check_score(self) -> FewshotExampleMultiLabel:
    """Validate that scores lie within [0, 1].

    :return: Validated instance.
    """
    if any([conf for conf in self.score_per_label.values() if not 0 <= conf <= 1]):
        raise ValueError("Score has to be between 0 and 1.")
    return self

from_dspy(example) classmethod

Convert from dspy.Example.

Parameters:

Name	Type	Description	Default
example	Example	Example as dspy.Example.	required

Returns:

Type	Description
Self	Example as FewshotExample.

Source code in sieves/tasks/predictive/schemas/core.py

@classmethod
def from_dspy(cls, example: dspy.Example) -> Self:
    """Convert from `dspy.Example`.

    :param example: Example as `dspy.Example`.
    :returns: Example as `FewshotExample`.
    """
    return cls(**example)

to_dspy()

Convert to dspy.Example.

Returns:

Type	Description
Example	Example as dspy.Example.

Source code in sieves/tasks/predictive/schemas/core.py

def to_dspy(self) -> dspy.Example:
    """Convert to `dspy.Example`.

    :returns: Example as `dspy.Example`.
    """
    return dspy.Example(**ModelWrapper.convert_fewshot_examples([self])[0]).with_inputs(*self.input_fields)

FewshotExampleSingleLabel

Bases: FewshotExample

Few‑shot example for single‑label classification with a global score.

Attributes: text: Input text. label: Predicted label. score: Confidence score.

Source code in sieves/tasks/predictive/schemas/classification.py

class FewshotExampleSingleLabel(BaseFewshotExample):
    """Few‑shot example for single‑label classification with a global score.

    Attributes:
        text: Input text.
        label: Predicted label.
        score: Confidence score.
    """

    label: str
    score: float

    @pydantic.model_validator(mode="after")
    def check_score(self) -> FewshotExampleSingleLabel:
        """Check score value.

        :return: Validated instance.
        """
        if not (0 <= self.score <= 1):
            raise ValueError("Score has to be between 0 and 1.")
        return self

    @property
    def target_fields(self) -> tuple[str, ...]:
        """Return target fields.

        :return: Target fields.
        """
        return ("label", "score")

input_fields property

Defines which fields are inputs.

Returns:

Type	Description
Sequence[str]	Sequence of field names.

target_fields property

Return target fields.

Returns:

Type	Description
tuple[str, ...]	Target fields.

check_score()

Check score value.

Returns:

Type	Description
FewshotExampleSingleLabel	Validated instance.

Source code in sieves/tasks/predictive/schemas/classification.py

@pydantic.model_validator(mode="after")
def check_score(self) -> FewshotExampleSingleLabel:
    """Check score value.

    :return: Validated instance.
    """
    if not (0 <= self.score <= 1):
        raise ValueError("Score has to be between 0 and 1.")
    return self

from_dspy(example) classmethod

Convert from dspy.Example.

Parameters:

Name	Type	Description	Default
example	Example	Example as dspy.Example.	required

Returns:

Type	Description
Self	Example as FewshotExample.

Source code in sieves/tasks/predictive/schemas/core.py

@classmethod
def from_dspy(cls, example: dspy.Example) -> Self:
    """Convert from `dspy.Example`.

    :param example: Example as `dspy.Example`.
    :returns: Example as `FewshotExample`.
    """
    return cls(**example)

to_dspy()

Convert to dspy.Example.

Returns:

Type	Description
Example	Example as dspy.Example.

Source code in sieves/tasks/predictive/schemas/core.py

def to_dspy(self) -> dspy.Example:
    """Convert to `dspy.Example`.

    :returns: Example as `dspy.Example`.
    """
    return dspy.Example(**ModelWrapper.convert_fewshot_examples([self])[0]).with_inputs(*self.input_fields)

ResultMultiLabel

Bases: BaseModel

Result of a multi-label classification task.

Attributes: label_scores: List of label-score pairs.

Source code in sieves/tasks/predictive/schemas/classification.py

class ResultMultiLabel(pydantic.BaseModel):
    """Result of a multi-label classification task.

    Attributes:
        label_scores: List of label-score pairs.
    """

    label_scores: list[tuple[str, float]]

ResultSingleLabel

Bases: BaseModel

Result of a single-label classification task.

Attributes: label: Predicted label. score: Confidence score.

Source code in sieves/tasks/predictive/schemas/classification.py

class ResultSingleLabel(pydantic.BaseModel):
    """Result of a single-label classification task.

    Attributes:
        label: Predicted label.
        score: Confidence score.
    """

    label: str
    score: float