Information Extraction

The InformationExtraction task allows for structured data extraction from documents using Pydantic schemas.

Usage

Confidence scores

Confidence scores are automatically captured and visible to the model (including nested fields in DSPy signatures). While few-shot examples are helpful for performance tuning, they are not required for the model to understand it should provide scores.

Multi-Entity Extraction (Default)

By default, the task operates in mode="multi", finding all instances of the specified entity.

import pydantic
from sieves import tasks
from sieves.tasks.predictive.information_extraction import FewshotExampleMulti

class Person(pydantic.BaseModel, frozen=True):
    name: str
    age: int

examples = [
    FewshotExampleMulti(
        text="Alice is 30 and Bob is 25.",
        entities=[Person(name="Alice", age=30), Person(name="Bob", age=25)]
    )
]

task = tasks.InformationExtraction(
    entity_type=Person,
    mode="multi",
    fewshot_examples=examples,
    model=model,
)

Single-Entity Extraction

Use mode="single" when you expect exactly one entity per document (or none). This is useful for summarizing a document into a structured record.

from sieves import tasks
from sieves.tasks.predictive.information_extraction import FewshotExampleSingle

class Invoice(pydantic.BaseModel, frozen=True):
    id: str
    total: float

examples = [
    FewshotExampleSingle(
        text="Invoice #123: $50.00",
        entity=Invoice(id="123", total=50.0)
    )
]

task = tasks.InformationExtraction(
    entity_type=Invoice,
    mode="single",
    fewshot_examples=examples,
    model=model,
)

Results

The InformationExtraction task produces unified results based on the chosen mode:

class ResultSingle(pydantic.BaseModel):
    """Result of a single-entity extraction task.

    Attributes:
        entity: Extracted entity.
    """

    entity: pydantic.BaseModel | None


class ResultMulti(pydantic.BaseModel):
    """Result of a multi-entity extraction task.

    Attributes:
        entities: List of extracted entities.
    """

    entities: list[pydantic.BaseModel]

• mode="multi": Returns a ResultMulti object with an entities list.
• mode="single": Returns a ResultSingle object with a single entity (or None).

Confidence Scores

To provide confidence scores for user-defined entity types, sieves automatically creates a subclass of your provided Pydantic model that includes a score field.

The instances returned in the results will have this additional attribute:

class MyEntity(pydantic.BaseModel, frozen=True):
    name: str

# ... execution ...

result = doc.results["my_task"].entity
print(result.name)
print(result.score)  # Confidence score between 0 and 1, or None for some LLM outputs

While confidence scores are always present for GLiNER2 models, they are self-reported and optional for LLMs (DSPy, Outlines, LangChain).

If your original model already contains a score field, sieves will use it as-is without further modification.

Evaluation

The performance of information extraction can be measured using the .evaluate() method.

• Metric:
  • Single Mode: Accuracy (Accuracy). The fraction of documents where the extracted entity exactly matches the ground truth.
  • Multi Mode: Corpus-wide Micro-F1 Score (F1). True Positives, False Positives, and False Negatives are accumulated across all documents based on exact entity matches.
• Requirement: Each document must have ground-truth entities (matching your schema) stored in doc.gold[task_id].

report = task.evaluate(docs)
# Use report.metrics['F1'] or report.metrics['Accuracy'] depending on mode
print(f"IE Score: {report.metrics.get('F1') or report.metrics.get('Accuracy')}")

Ground Truth Formats

Ground truth has to be specified in doc.gold using ResultMulti or ResultSingle instances.

---

Information extraction.

InformationExtraction

Bases: PredictiveTask[TaskPromptSignature, TaskResult, _TaskBridge]

Information extraction task.

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

class InformationExtraction(PredictiveTask[TaskPromptSignature, TaskResult, _TaskBridge]):
    """Information extraction task."""

    def __init__(
        self,
        entity_type: type[pydantic.BaseModel],
        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["multi", "single"] = "multi",
        model_settings: ModelSettings = ModelSettings(),
        condition: Callable[[Doc], bool] | None = None,
    ) -> None:
        """Initialize new PredictiveTask.

        :param entity_type: Pydantic model class representing the object type to extract.
        :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: Extraction mode. If "multi", all occurrences of the entity are extracted. If "single", exactly one
            (or no) entity is extracted.
        :param model_settings: Settings for structured generation.
        :param condition: Optional callable that determines whether to process each document.
        """
        self._entity_type = entity_type
        self._mode = mode
        self._scored_entity_type = _wrap_with_score(entity_type)

        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,
        )

        if (
            isinstance(entity_type, type)
            and issubclass(entity_type, pydantic.BaseModel)
            and not entity_type.model_config.get("frozen", False)
        ):
            raise ValueError(
                f"Entity type provided to task {self._task_id} isn't frozen, which means that entities can't "
                f"be deduplicated and compared. Modify entity_type to be frozen=True."
            )

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

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

    @property
    @override
    def prompt_signature(self) -> type[pydantic.BaseModel]:
        """Return the unified Pydantic prompt signature for this task.

        :return: Unified Pydantic prompt signature.
        """
        scored_type = self._scored_entity_type

        if self._mode == "multi":
            return pydantic.create_model(
                f"{scored_type.__name__}Multi",
                __doc__=f"Result of multi-entity extraction for {scored_type.__name__}.",
                entities=(
                    list[scored_type],  # type: ignore[valid-type]
                    pydantic.Field(..., description=f"List of extracted {scored_type.__name__} entities."),
                ),
            )
        return pydantic.create_model(
            f"{scored_type.__name__}Single",
            __doc__=f"Result of single-entity extraction for {scored_type.__name__}.",
            entity=(
                scored_type | None,  # type: ignore[valid-type]
                pydantic.Field(..., description=f"The extracted {scored_type.__name__} entity."),
            ),
        )

    @property
    @override
    def metric(self) -> str:
        return "F1" if self._mode == "multi" else "Accuracy"

    @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.
        """
        # Single mode.
        if self._mode == "single":
            correct = 0
            total = 0
            for gold, pred in zip(truths, preds):
                # If gold is None, we assume it's a negative example (no entity).
                # If pred is None, it predicted no entity.
                # If both None -> Correct.
                if gold is not None:
                    assert isinstance(gold, TaskResult)
                    gold_entity = gold.entity
                else:
                    gold_entity = None

                if pred is not None:
                    assert isinstance(pred, TaskResult)
                    pred_entity = pred.entity
                else:
                    pred_entity = None

                if gold_entity == pred_entity:
                    correct += 1
                total += 1

            return {self.metric: correct / total if total > 0 else 0.0}

        # Multi mode.
        tp = 0
        fp = 0
        fn = 0

        def entity_to_tuple(entity: Any) -> tuple:
            # Handle Pydantic model
            d = entity.model_dump()

            # Remove score if present
            if "score" in d:
                del d["score"]

            items = sorted(d.items())
            return tuple((k, v if not (isinstance(v, list) or isinstance(v, dict)) else str(v)) for k, v in items)

        for gold, pred in zip(truths, preds):
            if gold is not None:
                assert isinstance(gold, TaskResult)
                true_entities = {entity_to_tuple(e) for e in gold.entities}
            else:
                true_entities = set()

            if pred is not None:
                assert isinstance(pred, TaskResult)
                pred_entities = {entity_to_tuple(e) for e in pred.entities}
            else:
                pred_entities = set()

            tp += len(true_entities & pred_entities)
            fp += len(pred_entities - true_entities)
            fn += len(true_entities - pred_entities)

        precision = tp / (tp + fp) if (tp + fp) > 0 else 0.0
        recall = tp / (tp + fn) if (tp + fn) > 0 else 0.0
        f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0.0

        return {self.metric: f1}

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

        :param model_type: Type of model to initialize bridge for.
        :return _TaskBridge: ModelWrapper task bridge.
        :raises ValueError: If model type is not supported.
        :raises TypeError: On entity type and model type mismatch.
        """
        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.structure,
                mode=self._mode,
            )

        bridge_types: dict[ModelType, type[_TaskBridge]] = {
            ModelType.dspy: DSPyInformationExtraction,
            ModelType.langchain: PydanticInformationExtraction,
            ModelType.outlines: PydanticInformationExtraction,
        }

        try:
            bridge = bridge_types[model_type](
                task_id=self._task_id,
                prompt_instructions=self._custom_prompt_instructions,
                entity_type=self._scored_entity_type,
                model_settings=self._model_settings,
                mode=self._mode,
                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

        return bridge

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

    @override
    @property
    def _state(self) -> dict[str, Any]:
        return {
            **super()._state,
            "entity_type": self._entity_type,
            "mode": self._mode,
        }

    @override
    def _validate_fewshot_examples(self) -> None:
        example_type_error_text = "Fewshot example type mismatch: mode = {mode} requires {example_type}."

        for i, fs_example in enumerate(self._fewshot_examples or []):
            if self._mode == "multi":
                assert isinstance(fs_example, FewshotExampleMulti), TypeError(
                    example_type_error_text.format(example_type=FewshotExampleMulti, mode=self._mode)
                )
            else:
                assert isinstance(fs_example, FewshotExampleSingle), TypeError(
                    example_type_error_text.format(example_type=FewshotExampleSingle, mode=self._mode)
                )

    @override
    def to_hf_dataset(self, docs: Iterable[Doc], threshold: float | None = None) -> datasets.Dataset:
        # Use the scored entity type for the dataset schema.
        # We need to unwrap the unified signature container if it exists.
        entity_type = self.prompt_signature
        if "entities" in entity_type.model_fields:
            entity_type = entity_type.model_fields["entities"].annotation
            if typing.get_origin(entity_type) is list:
                entity_type = typing.get_args(entity_type)[0]
        elif "entity" in entity_type.model_fields:
            entity_type = entity_type.model_fields["entity"].annotation
            if typing.get_origin(entity_type) in (typing.Union, types.UnionType):
                args = typing.get_args(entity_type)
                entity_type = [t for t in args if t is not type(None)][0]

        # Define metadata.
        hf_entity_type = PydanticToHFDatasets.model_cls_to_features(entity_type)
        if self._mode == "multi":
            features = datasets.Features(
                {"text": datasets.Value("string"), "entities": datasets.Sequence(hf_entity_type)}
            )
        else:
            features = datasets.Features({"text": datasets.Value("string"), "entity": hf_entity_type})

        entity_type_name = getattr(entity_type, "__name__", str(entity_type))
        info = datasets.DatasetInfo(
            description=f"Information extraction dataset for entity type {entity_type_name}. "
            f"Generated with sieves v{Config.get_version()}.",
            features=features,
        )

        # Fetch data used for generating dataset.
        try:
            if self._mode == "multi":
                data = [
                    (doc.text, [PydanticToHFDatasets.model_to_dict(res) for res in doc.results[self._task_id].entities])
                    for doc in docs
                ]
            else:
                data = [
                    (
                        doc.text,
                        PydanticToHFDatasets.model_to_dict(doc.results[self._task_id].entity)
                        if doc.results[self._task_id].entity
                        else None,
                    )
                    for doc in docs
                ]
        except KeyError as err:
            raise KeyError(f"Not all documents have results for this task with ID {self._task_id}") from err

        # Create dataset.
        return datasets.Dataset.from_list(
            [
                {
                    "text": text,
                    ("entities" if self._mode == "multi" else "entity"): res,
                }
                for text, res in data
            ],
            features=features,
            info=info,
        )

    @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:
        raise NotImplementedError

    @override
    def _evaluate_dspy_example(self, truth: dspy.Example, pred: dspy.Prediction, trace: Any, model: dspy.LM) -> float:
        if self._mode == "single":
            true_entity = truth.get("entity")
            pred_entity = pred.get("entity")
            base_accuracy = 1.0 if true_entity == pred_entity else 0.0

            # If score is available in both truth and pred, incorporate it into the metric.
            if "score" in truth and truth["score"] is not None and "score" in pred and pred["score"] is not None:
                score_accuracy = 1 - abs(truth["score"] - max(min(pred["score"], 1), 0))
                return (base_accuracy + score_accuracy) / 2

            return base_accuracy

        def entity_to_tuple(entity: dict) -> tuple:
            """Convert entity dict to hashable tuple for comparison.

            Converts nested structures (lists, dicts) to strings to make them hashable.

            :param entity: Entity dictionary to convert.
            :return: Hashable tuple representation of the entity.
            """
            items = sorted(entity.items())
            return tuple((k, v if not (isinstance(v, list) or isinstance(v, dict)) else str(v)) for k, v in items)

        # Compute set-based F1 score for entity extraction
        true_entities = {entity_to_tuple(e) for e in truth["entities"]}
        pred_entities = {entity_to_tuple(e) for e in pred.get("entities", [])}

        if not true_entities:
            return 1.0 if not pred_entities else 0.0

        precision = len(true_entities & pred_entities) / len(pred_entities) if pred_entities else 0
        recall = len(true_entities & pred_entities) / len(true_entities)
        f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0

        # If scores are available, incorporate them.
        if "scores" in truth and truth["scores"] is not None and "scores" in pred and pred["scores"] is not None:
            # Simple averaging of score similarities for matched entities is complex here due to set-based matching.
            # We'll just average the overall entity f1 with a score similarity metric.
            true_avg_score = sum(truth["scores"]) / len(truth["scores"]) if truth["scores"] else 0
            pred_avg_score = sum(pred["scores"]) / len(pred["scores"]) if pred["scores"] else 0
            score_accuracy = 1 - abs(true_avg_score - max(min(pred_avg_score, 1), 0))
            return (f1 + score_accuracy) / 2

        return f1

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 property

Return the unified Pydantic prompt signature for this task.

Returns:

Type	Description
type[BaseModel]	Unified Pydantic prompt signature.

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__(entity_type, model, task_id=None, include_meta=True, batch_size=-1, prompt_instructions=None, fewshot_examples=(), mode='multi', model_settings=ModelSettings(), condition=None)

Initialize new PredictiveTask.

Parameters:

Name	Type	Description	Default
entity_type	type[BaseModel]	Pydantic model class representing the object type to extract.	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['multi', 'single']	Extraction mode. If "multi", all occurrences of the entity are extracted. If "single", exactly one (or no) entity is extracted.	'multi'
model_settings	ModelSettings	Settings for structured generation.	ModelSettings()
condition	Callable[[Doc], bool] | None	Optional callable that determines whether to process each document.	None

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

def __init__(
    self,
    entity_type: type[pydantic.BaseModel],
    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["multi", "single"] = "multi",
    model_settings: ModelSettings = ModelSettings(),
    condition: Callable[[Doc], bool] | None = None,
) -> None:
    """Initialize new PredictiveTask.

    :param entity_type: Pydantic model class representing the object type to extract.
    :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: Extraction mode. If "multi", all occurrences of the entity are extracted. If "single", exactly one
        (or no) entity is extracted.
    :param model_settings: Settings for structured generation.
    :param condition: Optional callable that determines whether to process each document.
    """
    self._entity_type = entity_type
    self._mode = mode
    self._scored_entity_type = _wrap_with_score(entity_type)

    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,
    )

    if (
        isinstance(entity_type, type)
        and issubclass(entity_type, pydantic.BaseModel)
        and not entity_type.model_config.get("frozen", False)
    ):
        raise ValueError(
            f"Entity type provided to task {self._task_id} isn't frozen, which means that entities can't "
            f"be deduplicated and compared. Modify entity_type to be frozen=True."
        )

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()})

Bridges for information extraction task.

DSPyInformationExtraction

Bases: InformationExtractionBridge[PromptSignature, Result, InferenceMode]

DSPy bridge for information extraction.

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

class DSPyInformationExtraction(InformationExtractionBridge[dspy_.PromptSignature, dspy_.Result, dspy_.InferenceMode]):
    """DSPy bridge for information extraction."""

    @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

    @override
    def integrate(self, results: Sequence[dspy_.Result], docs: list[Doc]) -> list[Doc]:
        for doc, result in zip(docs, results):
            if self._mode == "multi":
                assert len(result.completions.entities) == 1
                doc.results[self._task_id] = ResultMulti(
                    entities=result.completions.entities[0],
                )
            else:
                assert len(result.completions.entity) == 1
                doc.results[self._task_id] = ResultSingle(
                    entity=result.completions.entity[0],
                )
        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 res_clean in consolidated_results_clean:
            if self._mode == "multi":
                consolidated_results.append(
                    dspy.Prediction.from_completions(
                        {"entities": [res_clean]},
                        signature=self.prompt_signature,
                    )
                )
            else:
                consolidated_results.append(
                    dspy.Prediction.from_completions(
                        {"entity": [res_clean]},
                        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_signature property

Create output signature.

E.g.: Signature in DSPy, Pydantic objects in outlines, JSON schema in jsonformers. This is model type-specific.

Returns:

Type	Description
type[TaskPromptSignature] | TaskPromptSignature	Output signature object. This can be an instance (e.g. a regex string) or a class (e.g. a Pydantic class).

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, entity_type, model_settings, mode, prompt_signature, model_type, fewshot_examples=())

Initialize information extraction bridge.

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
entity_type	type[BaseModel]	Object type to extract.	required
model_settings	ModelSettings	Settings for structured generation.	required
mode	Literal['multi', 'single']	Extraction mode. If "multi", all occurrences of the entity are extracted. If "single", exactly one (or no) entity is extracted.	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/information_extraction/bridges.py

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

    :param task_id: Task ID.
    :param prompt_instructions: Custom prompt instructions. If None, default instructions are used.
    :param entity_type: Object type to extract.
    :param model_settings: Settings for structured generation.
    :param mode: Extraction mode. If "multi", all occurrences of the entity are extracted. If "single", exactly one
        (or no) entity is extracted.
    :param prompt_signature: Unified Pydantic prompt signature.
    :param model_type: Model type.
    :param fewshot_examples: Few-shot examples.
    """
    assert model_type in {ModelType.dspy, ModelType.gliner, ModelType.langchain, ModelType.outlines}

    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,
    )
    self._entity_type = entity_type
    self._mode = mode

    if self._mode == "multi":
        self._consolidation_strategy = MultiEntityConsolidation(extractor=self._get_multi_extractor())
    else:
        self._consolidation_strategy = SingleEntityConsolidation(extractor=self._get_single_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]

InformationExtractionBridge

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

Abstract base class for information extraction bridges.

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

class InformationExtractionBridge(Bridge[_BridgePromptSignature, _BridgeResult, ModelWrapperInferenceMode], abc.ABC):
    """Abstract base class for information extraction bridges."""

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

        :param task_id: Task ID.
        :param prompt_instructions: Custom prompt instructions. If None, default instructions are used.
        :param entity_type: Object type to extract.
        :param model_settings: Settings for structured generation.
        :param mode: Extraction mode. If "multi", all occurrences of the entity are extracted. If "single", exactly one
            (or no) entity is extracted.
        :param prompt_signature: Unified Pydantic prompt signature.
        :param model_type: Model type.
        :param fewshot_examples: Few-shot examples.
        """
        assert model_type in {ModelType.dspy, ModelType.gliner, ModelType.langchain, ModelType.outlines}

        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,
        )
        self._entity_type = entity_type
        self._mode = mode

        if self._mode == "multi":
            self._consolidation_strategy = MultiEntityConsolidation(extractor=self._get_multi_extractor())
        else:
            self._consolidation_strategy = SingleEntityConsolidation(extractor=self._get_single_extractor())

    @staticmethod
    def _get_multi_extractor() -> Callable[[Any], Iterable[pydantic.BaseModel]]:
        """Return a callable that extracts a list of entities from a raw chunk result.

        :return: Multi-extractor callable.
        """
        return lambda res: res.entities

    @staticmethod
    def _get_single_extractor() -> Callable[[Any], pydantic.BaseModel | None]:
        """Return a callable that extracts a single entity from a raw chunk result.

        :return: Single-extractor callable.
        """
        return lambda res: res.entity

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_signature property

Create output signature.

E.g.: Signature in DSPy, Pydantic objects in outlines, JSON schema in jsonformers. This is model type-specific.

Returns:

Type	Description
type[TaskPromptSignature] | TaskPromptSignature	Output signature object. This can be an instance (e.g. a regex string) or a class (e.g. a Pydantic class).

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, entity_type, model_settings, mode, prompt_signature, model_type, fewshot_examples=())

Initialize information extraction bridge.

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
entity_type	type[BaseModel]	Object type to extract.	required
model_settings	ModelSettings	Settings for structured generation.	required
mode	Literal['multi', 'single']	Extraction mode. If "multi", all occurrences of the entity are extracted. If "single", exactly one (or no) entity is extracted.	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/information_extraction/bridges.py

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

    :param task_id: Task ID.
    :param prompt_instructions: Custom prompt instructions. If None, default instructions are used.
    :param entity_type: Object type to extract.
    :param model_settings: Settings for structured generation.
    :param mode: Extraction mode. If "multi", all occurrences of the entity are extracted. If "single", exactly one
        (or no) entity is extracted.
    :param prompt_signature: Unified Pydantic prompt signature.
    :param model_type: Model type.
    :param fewshot_examples: Few-shot examples.
    """
    assert model_type in {ModelType.dspy, ModelType.gliner, ModelType.langchain, ModelType.outlines}

    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,
    )
    self._entity_type = entity_type
    self._mode = mode

    if self._mode == "multi":
        self._consolidation_strategy = MultiEntityConsolidation(extractor=self._get_multi_extractor())
    else:
        self._consolidation_strategy = SingleEntityConsolidation(extractor=self._get_single_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.
    """

PydanticInformationExtraction

Bases: InformationExtractionBridge[BaseModel, BaseModel, ModelWrapperInferenceMode]

Base class for Pydantic-based information extraction bridges.

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

class PydanticInformationExtraction(
    InformationExtractionBridge[pydantic.BaseModel, pydantic.BaseModel, ModelWrapperInferenceMode],
):
    """Base class for Pydantic-based information extraction bridges."""

    @override
    def _validate(self) -> None:
        assert self._model_type in {ModelType.langchain, ModelType.outlines}

    @override
    @property
    def _default_prompt_instructions(self) -> str:
        if self._mode == "multi":
            return (
                "Find all occurences of this kind of entitity within the text. For each entity found, also provide "
                "a confidence score between 0.0 and 1.0 in the 'score' field."
            )

        return (
            "Find the single most relevant entitity within the text. If no such entitity exists, return null. "
            "Return exactly one entity with all its fields, NOT just a string. Also provide a confidence score "
            "between 0.0 and 1.0 in the 'score' field."
        )

    @override
    @property
    def _prompt_conclusion(self) -> str | None:
        return "========\n\n<text>{{ text }}</text>"

    @override
    def integrate(self, results: Sequence[pydantic.BaseModel], docs: list[Doc]) -> list[Doc]:
        for doc, result in zip(docs, results):
            if self._mode == "multi":
                assert hasattr(result, "entities")
                doc.results[self._task_id] = ResultMulti(entities=result.entities)
            else:
                assert hasattr(result, "entity")
                doc.results[self._task_id] = ResultSingle(entity=result.entity)
        return docs

    @override
    def consolidate(
        self,
        results: Sequence[pydantic.BaseModel],
        docs_offsets: list[tuple[int, int]],
    ) -> Sequence[pydantic.BaseModel]:
        assert issubclass(self.prompt_signature, pydantic.BaseModel)

        consolidated_results_clean = self._consolidation_strategy.consolidate(results, docs_offsets)
        consolidated_results: list[pydantic.BaseModel] = []

        for res_clean in consolidated_results_clean:
            if self._mode == "multi":
                consolidated_results.append(self.prompt_signature(entities=res_clean))
            else:
                consolidated_results.append(self.prompt_signature(entity=res_clean))

        return consolidated_results

    @override
    @property
    def inference_mode(self) -> outlines_.InferenceMode | langchain_.InferenceMode:
        if self._model_type == ModelType.outlines:
            return self._model_settings.inference_mode or outlines_.InferenceMode.json

        elif self._model_type == ModelType.langchain:
            return self._model_settings.inference_mode or langchain_.InferenceMode.structured

        raise ValueError(f"Unsupported model type: {self._model_type}")

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_signature property

Create output signature.

E.g.: Signature in DSPy, Pydantic objects in outlines, JSON schema in jsonformers. This is model type-specific.

Returns:

Type	Description
type[TaskPromptSignature] | TaskPromptSignature	Output signature object. This can be an instance (e.g. a regex string) or a class (e.g. a Pydantic class).

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, entity_type, model_settings, mode, prompt_signature, model_type, fewshot_examples=())

Initialize information extraction bridge.

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
entity_type	type[BaseModel]	Object type to extract.	required
model_settings	ModelSettings	Settings for structured generation.	required
mode	Literal['multi', 'single']	Extraction mode. If "multi", all occurrences of the entity are extracted. If "single", exactly one (or no) entity is extracted.	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/information_extraction/bridges.py

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

    :param task_id: Task ID.
    :param prompt_instructions: Custom prompt instructions. If None, default instructions are used.
    :param entity_type: Object type to extract.
    :param model_settings: Settings for structured generation.
    :param mode: Extraction mode. If "multi", all occurrences of the entity are extracted. If "single", exactly one
        (or no) entity is extracted.
    :param prompt_signature: Unified Pydantic prompt signature.
    :param model_type: Model type.
    :param fewshot_examples: Few-shot examples.
    """
    assert model_type in {ModelType.dspy, ModelType.gliner, ModelType.langchain, ModelType.outlines}

    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,
    )
    self._entity_type = entity_type
    self._mode = mode

    if self._mode == "multi":
        self._consolidation_strategy = MultiEntityConsolidation(extractor=self._get_multi_extractor())
    else:
        self._consolidation_strategy = SingleEntityConsolidation(extractor=self._get_single_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 information extraction task.

FewshotExampleMulti

Bases: FewshotExample

Few-shot example for multi-entity extraction.

Attributes: text: Input text. entities: List of entities.

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

class FewshotExampleMulti(BaseFewshotExample):
    """Few-shot example for multi-entity extraction.

    Attributes:
        text: Input text.
        entities: List of entities.
    """

    entities: list[pydantic.BaseModel]

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

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

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.

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)

FewshotExampleSingle

Bases: FewshotExample

Few-shot example for single-entity extraction.

Attributes: text: Input text. entity: Extracted entity.

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

class FewshotExampleSingle(BaseFewshotExample):
    """Few-shot example for single-entity extraction.

    Attributes:
        text: Input text.
        entity: Extracted entity.
    """

    entity: pydantic.BaseModel | None

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

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

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.

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)

ResultMulti

Bases: BaseModel

Result of a multi-entity extraction task.

Attributes: entities: List of extracted entities.

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

class ResultMulti(pydantic.BaseModel):
    """Result of a multi-entity extraction task.

    Attributes:
        entities: List of extracted entities.
    """

    entities: list[pydantic.BaseModel]

ResultSingle

Bases: BaseModel

Result of a single-entity extraction task.

Attributes: entity: Extracted entity.

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

class ResultSingle(pydantic.BaseModel):
    """Result of a single-entity extraction task.

    Attributes:
        entity: Extracted entity.
    """

    entity: pydantic.BaseModel | None