Summarization

Text summarization predictive task.

FewshotExample

Bases: FewshotExample

Few-shot example with a target summary.

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

class FewshotExample(BaseFewshotExample):
    """Few-shot example with a target summary."""

    n_words: int
    summary: str

    @override
    @property
    def target_fields(self) -> Sequence[str]:
        return ("summary",)

input_fields property

Defines which fields are inputs.

Returns:

Type	Description
Sequence[str]	Sequence of field names.

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/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/core.py

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

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

Summarization

Bases: PredictiveTask[_TaskPromptSignature, _TaskResult, _TaskBridge]

Summarize documents to a target length using structured engines.

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

class Summarization(PredictiveTask[_TaskPromptSignature, _TaskResult, _TaskBridge]):
    """Summarize documents to a target length using structured engines."""

    def __init__(
        self,
        n_words: int,
        model: _TaskModel,
        task_id: str | None = None,
        include_meta: bool = True,
        batch_size: int = -1,
        overwrite: bool = False,
        prompt_instructions: str | None = None,
        fewshot_examples: Sequence[FewshotExample] = (),
        generation_settings: GenerationSettings = GenerationSettings(),
    ) -> None:
        """Initialize new Summarization task.

        :param n_words: Maximal number of words (consider this a guideline, not a strict limit).
        :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 overwrite: Some tasks, e.g. anonymization or translation, output a modified version of the input text.
            If True, these tasks overwrite the original document text. If False, the result will just be stored in the
            documents' `.results` field.
        :param prompt_instructions: Custom prompt instructions. If None, default instructions are used.
        :param fewshot_examples: Few-shot examples.
        :param generation_settings: Settings for structured generation.
        """
        self._n_words = n_words

        super().__init__(
            model=model,
            task_id=task_id,
            include_meta=include_meta,
            batch_size=batch_size,
            overwrite=overwrite,
            prompt_instructions=prompt_instructions,
            fewshot_examples=fewshot_examples,
            generation_settings=generation_settings,
        )

    @override
    def _init_bridge(self, engine_type: EngineType) -> _TaskBridge:
        if engine_type == EngineType.glix:
            return GliXBridge(
                task_id=self._task_id,
                prompt_instructions=self._custom_prompt_instructions,
                prompt_signature=[],
                inference_mode=glix_.InferenceMode.summarization,
            )

        bridge_types: dict[EngineType, type[_TaskBridge]] = {
            EngineType.dspy: DSPySummarization,
            EngineType.langchain: LangChainSummarization,
            EngineType.outlines: OutlinesSummarization,
        }

        try:
            bridge_type = bridge_types[engine_type]
            assert not issubclass(bridge_type, GliXBridge)

            return bridge_type(
                task_id=self._task_id,
                prompt_instructions=self._custom_prompt_instructions,
                overwrite=self._overwrite,
                n_words=self._n_words,
            )
        except KeyError as err:
            raise KeyError(f"Engine type {engine_type} is not supported by {self.__class__.__name__}.") from err

    @property
    @override
    def supports(self) -> set[EngineType]:
        return {
            EngineType.dspy,
            EngineType.glix,
            EngineType.langchain,
            EngineType.outlines,
        }

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

    @override
    def to_hf_dataset(self, docs: Iterable[Doc], threshold: float = 0.5) -> datasets.Dataset:
        # Define metadata.
        features = datasets.Features({"text": datasets.Value("string"), "summary": datasets.Value("string")})
        info = datasets.DatasetInfo(
            description=f"Summarization dataset. Generated with sieves v{Config.get_version()}.",
            features=features,
        )

        # Fetch data used for generating dataset.
        try:
            data = [(doc.text, doc.results[self._task_id]) 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

        def generate_data() -> Iterable[dict[str, Any]]:
            """Yield results as dicts.

            :return: Results as dicts.
            """
            for text, summary in data:
                yield {"text": text, "summary": summary}

        # Create dataset.
        return datasets.Dataset.from_generator(generate_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

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 the task on a set of documents.

Parameters:

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

Returns:

Type	Description
Iterable[Doc]	Processed documents.

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

def __call__(self, docs: Iterable[Doc]) -> Iterable[Doc]:
    """Execute the task on a set of documents.

    :param docs: Documents to process.
    :return Iterable[Doc]: Processed documents.
    """
    # 1. Compile expected prompt signatures.
    signature = self._bridge.prompt_signature

    # 2. Build executable.
    executable = self._engine.build_executable(
        inference_mode=self._bridge.inference_mode,
        prompt_template=self.prompt_template,
        prompt_signature=signature,
        fewshot_examples=self._fewshot_examples,
    )

    # Compute batch-wise results.
    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)]:
        if len(docs_batch) == 0:
            break

        # 3. Extract values from docs to inject/render those into prompt templates.
        docs_values = list(self._bridge.extract(docs_batch))
        assert len(docs_values) == len(docs_batch)

        # 4. Map extracted docs values onto chunks.
        docs_chunks_offsets: list[tuple[int, int]] = []
        docs_chunks: list[dict[str, Any]] = []
        for doc, doc_values in zip(docs_batch, docs_values):
            assert doc.text
            doc_chunks_values = [doc_values | {"text": chunk} for chunk in (doc.chunks or [doc.text])]
            docs_chunks_offsets.append((len(docs_chunks), len(docs_chunks) + len(doc_chunks_values)))
            docs_chunks.extend(doc_chunks_values)

        # 5. Execute prompts per chunk.
        results = list(executable(docs_chunks))
        assert len(results) == len(docs_chunks)

        # 6. Consolidate chunk results.
        results = list(self._bridge.consolidate(results, docs_chunks_offsets))
        assert len(results) == len(docs_batch)

        # 7. Integrate results into docs.
        docs_batch = self._bridge.integrate(results, docs_batch)

        yield from docs_batch

__init__(n_words, model, task_id=None, include_meta=True, batch_size=-1, overwrite=False, prompt_instructions=None, fewshot_examples=(), generation_settings=GenerationSettings())

Initialize new Summarization task.

Parameters:

Name	Type	Description	Default
n_words	int	Maximal number of words (consider this a guideline, not a strict limit).	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
overwrite	bool	Some tasks, e.g. anonymization or translation, output a modified version of the input text. If True, these tasks overwrite the original document text. If False, the result will just be stored in the documents' .results field.	False
prompt_instructions	str | None	Custom prompt instructions. If None, default instructions are used.	None
fewshot_examples	Sequence[FewshotExample]	Few-shot examples.	()
generation_settings	GenerationSettings	Settings for structured generation.	GenerationSettings()

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

def __init__(
    self,
    n_words: int,
    model: _TaskModel,
    task_id: str | None = None,
    include_meta: bool = True,
    batch_size: int = -1,
    overwrite: bool = False,
    prompt_instructions: str | None = None,
    fewshot_examples: Sequence[FewshotExample] = (),
    generation_settings: GenerationSettings = GenerationSettings(),
) -> None:
    """Initialize new Summarization task.

    :param n_words: Maximal number of words (consider this a guideline, not a strict limit).
    :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 overwrite: Some tasks, e.g. anonymization or translation, output a modified version of the input text.
        If True, these tasks overwrite the original document text. If False, the result will just be stored in the
        documents' `.results` field.
    :param prompt_instructions: Custom prompt instructions. If None, default instructions are used.
    :param fewshot_examples: Few-shot examples.
    :param generation_settings: Settings for structured generation.
    """
    self._n_words = n_words

    super().__init__(
        model=model,
        task_id=task_id,
        include_meta=include_meta,
        batch_size=batch_size,
        overwrite=overwrite,
        prompt_instructions=prompt_instructions,
        fewshot_examples=fewshot_examples,
        generation_settings=generation_settings,
    )

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["generation_settings"] = GenerationSettings.model_validate(init_dict["generation_settings"])

    return cls(**init_dict)

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]
    pred_eval = functools.partial(self._evaluate_optimization_example, 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(EngineType.get_engine_type(self._engine))

    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 summarization task.

DSPySummarization

Bases: SummarizationBridge[PromptSignature, Result, InferenceMode]

DSPy bridge for summarization.

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

class DSPySummarization(SummarizationBridge[dspy_.PromptSignature, dspy_.Result, dspy_.InferenceMode]):
    """DSPy bridge for summarization."""

    @override
    @property
    def _default_prompt_instructions(self) -> str:
        return "Summary of a longer text."

    @override
    @property
    def _prompt_example_template(self) -> str | None:
        return None

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

    @override
    @cached_property
    def prompt_signature(self) -> type[dspy_.PromptSignature]:
        class Summary(dspy.Signature):  # type: ignore[misc]
            text: str = dspy.InputField(description="Text to summarize.")
            n_words: str = dspy.InputField(description="Number of words to approximately use for summary.")
            summary: str = dspy.OutputField(description="Summary of text.")

        Summary.__doc__ = jinja2.Template(self._prompt_instructions).render()

        return Summary

    @override
    @property
    def inference_mode(self) -> dspy_.InferenceMode:
        return dspy_.InferenceMode.chain_of_thought

    @override
    def integrate(self, results: Iterable[dspy_.Result], docs: Iterable[Doc]) -> Iterable[Doc]:
        for doc, result in zip(docs, results):
            assert len(result.completions.summary) == 1
            doc.results[self._task_id] = result.summary

            if self._overwrite:
                doc.text = result.summary

        return docs

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

        # Merge all chunk translations.
        for doc_offset in docs_offsets:
            summaries: list[str] = []

            for res in results[doc_offset[0] : doc_offset[1]]:
                if res is None:
                    continue
                summaries.append(res.summary)

            yield dspy.Prediction.from_completions(
                {"summary": ["\n".join(summaries)]},
                signature=self.prompt_signature,
            )

prompt_template property

Return prompt template.

Chains _prompt_instructions, _prompt_example_template and _prompt_conclusion.

Note: different engines have different expectations as 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 engine-specific expectations when creating a prompt template.

Returns:

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

__init__(task_id, prompt_instructions, overwrite, n_words)

Initialize InformationExtractionBridge.

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
overwrite	bool	Whether to overwrite text with summarization text.	required
n_words	int	Approximate number of words in summary.	required

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

def __init__(
    self,
    task_id: str,
    prompt_instructions: str | None,
    overwrite: bool,
    n_words: int,
):
    """Initialize InformationExtractionBridge.

    :param task_id: Task ID.
    :param prompt_instructions: Custom prompt instructions. If None, default instructions are used.
    :param overwrite: Whether to overwrite text with summarization text.
    :param n_words: Approximate number of words in summary.
    """
    super().__init__(
        task_id=task_id,
        prompt_instructions=prompt_instructions,
        overwrite=overwrite,
    )
    self._n_words = n_words

LangChainSummarization

Bases: PydanticBasedSummarization[InferenceMode]

LangChain bridge for summarization.

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

class LangChainSummarization(PydanticBasedSummarization[langchain_.InferenceMode]):
    """LangChain bridge for summarization."""

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

prompt_template property

Return prompt template.

Chains _prompt_instructions, _prompt_example_template and _prompt_conclusion.

Note: different engines have different expectations as 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 engine-specific expectations when creating a prompt template.

Returns:

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

__init__(task_id, prompt_instructions, overwrite, n_words)

Initialize InformationExtractionBridge.

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
overwrite	bool	Whether to overwrite text with summarization text.	required
n_words	int	Approximate number of words in summary.	required

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

def __init__(
    self,
    task_id: str,
    prompt_instructions: str | None,
    overwrite: bool,
    n_words: int,
):
    """Initialize InformationExtractionBridge.

    :param task_id: Task ID.
    :param prompt_instructions: Custom prompt instructions. If None, default instructions are used.
    :param overwrite: Whether to overwrite text with summarization text.
    :param n_words: Approximate number of words in summary.
    """
    super().__init__(
        task_id=task_id,
        prompt_instructions=prompt_instructions,
        overwrite=overwrite,
    )
    self._n_words = n_words

OutlinesSummarization

Bases: PydanticBasedSummarization[InferenceMode]

Outlines bridge for summarization.

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

class OutlinesSummarization(PydanticBasedSummarization[outlines_.InferenceMode]):
    """Outlines bridge for summarization."""

    @override
    @property
    def inference_mode(self) -> outlines_.InferenceMode:
        return outlines_.InferenceMode.json

prompt_template property

Return prompt template.

Chains _prompt_instructions, _prompt_example_template and _prompt_conclusion.

Note: different engines have different expectations as 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 engine-specific expectations when creating a prompt template.

Returns:

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

__init__(task_id, prompt_instructions, overwrite, n_words)

Initialize InformationExtractionBridge.

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
overwrite	bool	Whether to overwrite text with summarization text.	required
n_words	int	Approximate number of words in summary.	required

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

def __init__(
    self,
    task_id: str,
    prompt_instructions: str | None,
    overwrite: bool,
    n_words: int,
):
    """Initialize InformationExtractionBridge.

    :param task_id: Task ID.
    :param prompt_instructions: Custom prompt instructions. If None, default instructions are used.
    :param overwrite: Whether to overwrite text with summarization text.
    :param n_words: Approximate number of words in summary.
    """
    super().__init__(
        task_id=task_id,
        prompt_instructions=prompt_instructions,
        overwrite=overwrite,
    )
    self._n_words = n_words

PydanticBasedSummarization

Bases: SummarizationBridge[BaseModel, BaseModel, EngineInferenceMode], ABC

Base class for Pydantic-based summarization bridges.

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

class PydanticBasedSummarization(
    SummarizationBridge[pydantic.BaseModel, pydantic.BaseModel, EngineInferenceMode],
    abc.ABC,
):
    """Base class for Pydantic-based summarization bridges."""

    @override
    @property
    def _default_prompt_instructions(self) -> str:
        return """
        Your goal is to summarize a text. This summary should be around {{ max_n }} words.
        """

    @override
    @property
    def _prompt_example_template(self) -> str:
        return """
        {% if examples|length > 0 -%}
            <examples>
            {%- for example in examples %}
                <text>{{ example.text }}</text>
                <approximate_number_of_words_in_summary>{{ example.n_words }}</approximate_number_of_words_in_summary>
                <summary>
                {{ example.summary }}
                </summary>
            {% endfor -%}
            </examples>
        {% endif -%}
        """

    @override
    @property
    def _prompt_conclusion(self) -> str:
        return """
        ========
        <text>{{ text }}</text>
        <approximate_number_of_words_in_summary>{{ n_words }}</approximate_number_of_words_in_summary>
        <summary>
        """

    @override
    @cached_property
    def prompt_signature(self) -> type[pydantic.BaseModel]:
        class Summary(pydantic.BaseModel, frozen=True):
            """Summary of the specified text."""

            summary: str

        return Summary

    @override
    def integrate(self, results: Iterable[pydantic.BaseModel], docs: Iterable[Doc]) -> Iterable[Doc]:
        for doc, result in zip(docs, results):
            assert hasattr(result, "summary")
            doc.results[self._task_id] = result.summary

            if self._overwrite:
                doc.text = result.summary
        return docs

    @override
    def consolidate(
        self, results: Iterable[pydantic.BaseModel], docs_offsets: list[tuple[int, int]]
    ) -> Iterable[pydantic.BaseModel]:
        results = list(results)

        # Determine label scores for chunks per document.
        for doc_offset in docs_offsets:
            summaries: list[str] = []

            for res in results[doc_offset[0] : doc_offset[1]]:
                if res is None:
                    continue  # type: ignore[unreachable]

                assert hasattr(res, "summary")
                summaries.append(res.summary)

            yield self.prompt_signature(summary="\n".join(summaries).strip())

inference_mode abstractmethod property

Return inference mode.

Returns:

Type	Description
EngineInferenceMode	Inference mode.

prompt_template property

Return prompt template.

Chains _prompt_instructions, _prompt_example_template and _prompt_conclusion.

Note: different engines have different expectations as 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 engine-specific expectations when creating a prompt template.

Returns:

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

__init__(task_id, prompt_instructions, overwrite, n_words)

Initialize InformationExtractionBridge.

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
overwrite	bool	Whether to overwrite text with summarization text.	required
n_words	int	Approximate number of words in summary.	required

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

def __init__(
    self,
    task_id: str,
    prompt_instructions: str | None,
    overwrite: bool,
    n_words: int,
):
    """Initialize InformationExtractionBridge.

    :param task_id: Task ID.
    :param prompt_instructions: Custom prompt instructions. If None, default instructions are used.
    :param overwrite: Whether to overwrite text with summarization text.
    :param n_words: Approximate number of words in summary.
    """
    super().__init__(
        task_id=task_id,
        prompt_instructions=prompt_instructions,
        overwrite=overwrite,
    )
    self._n_words = n_words

SummarizationBridge

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

Abstract base class for summarization bridges.

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

class SummarizationBridge(
    Bridge[_BridgePromptSignature, _BridgeResult, EngineInferenceMode],
    abc.ABC,
):
    """Abstract base class for summarization bridges."""

    def __init__(
        self,
        task_id: str,
        prompt_instructions: str | None,
        overwrite: bool,
        n_words: int,
    ):
        """Initialize InformationExtractionBridge.

        :param task_id: Task ID.
        :param prompt_instructions: Custom prompt instructions. If None, default instructions are used.
        :param overwrite: Whether to overwrite text with summarization text.
        :param n_words: Approximate number of words in summary.
        """
        super().__init__(
            task_id=task_id,
            prompt_instructions=prompt_instructions,
            overwrite=overwrite,
        )
        self._n_words = n_words

    @override
    def extract(self, docs: Iterable[Doc]) -> Iterable[dict[str, Any]]:
        return ({"text": doc.text if doc.text else None, "n_words": self._n_words} for doc in docs)

inference_mode abstractmethod property

Return inference mode.

Returns:

Type	Description
EngineInferenceMode	Inference mode.

prompt_signature abstractmethod property

Create output signature.

E.g.: Signature in DSPy, Pydantic objects in outlines, JSON schema in jsonformers. This is engine-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_template and _prompt_conclusion.

Note: different engines have different expectations as 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 engine-specific expectations when creating a prompt template.

Returns:

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

__init__(task_id, prompt_instructions, overwrite, n_words)

Initialize InformationExtractionBridge.

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
overwrite	bool	Whether to overwrite text with summarization text.	required
n_words	int	Approximate number of words in summary.	required

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

def __init__(
    self,
    task_id: str,
    prompt_instructions: str | None,
    overwrite: bool,
    n_words: int,
):
    """Initialize InformationExtractionBridge.

    :param task_id: Task ID.
    :param prompt_instructions: Custom prompt instructions. If None, default instructions are used.
    :param overwrite: Whether to overwrite text with summarization text.
    :param n_words: Approximate number of words in summary.
    """
    super().__init__(
        task_id=task_id,
        prompt_instructions=prompt_instructions,
        overwrite=overwrite,
    )
    self._n_words = n_words

consolidate(results, docs_offsets) abstractmethod

Consolidate results for document chunks into document results.

Parameters:

Name	Type	Description	Default
results	Iterable[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
Iterable[TaskResult]	Results per document.

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

@abc.abstractmethod
def consolidate(self, results: Iterable[TaskResult], docs_offsets: list[tuple[int, int]]) -> Iterable[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 Iterable[_TaskResult]: Results per document.
    """

integrate(results, docs) abstractmethod

Integrate results into Doc instances.

Parameters:

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

Returns:

Type	Description
Iterable[Doc]	Updated doc instances.

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

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

    :param results: Results from prompt executable.
    :param docs: Doc instances to update.
    :return Iterable[Doc]: Updated doc instances.
    """