Chonkie

Allows chunking of documents into segments.

Chonkie

Bases: Task

Chunker wrapping the chonkie library.

Source code in sieves/tasks/preprocessing/chunking/chonkie_.py

class Chonkie(Task):
    """Chunker wrapping the chonkie library."""

    def __init__(
        self,
        chunker: chonkie.BaseChunker,
        task_id: str | None = None,
        include_meta: bool = False,
        batch_size: int = -1,
    ):
        """Initialize chunker.

        :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 processing. Use -1 to process all documents at once.
        """
        super().__init__(task_id=task_id, include_meta=include_meta, batch_size=batch_size)
        self._chunker = chunker

    def __call__(self, docs: Iterable[Doc]) -> Iterable[Doc]:
        """Split documents into chunks.

        :param docs: Documents to split.
        :return: Split documents.
        """
        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

            # Signature varies in between versions, try both.
            try:
                chunks = self._chunker.chunk_batch([doc.text for doc in docs_batch], show_progress=False)
            except TypeError:
                chunks = self._chunker.chunk_batch([doc.text for doc in docs_batch], show_progress_bar=False)  # type: ignore[unknown-argument]
            assert len(chunks) == len(docs_batch)

            for doc, doc_chunks in zip(docs_batch, chunks):
                if self._include_meta:
                    doc.meta |= {self.id: {doc_chunks}}
                doc.chunks = [chunk.text for chunk in doc_chunks]

                yield doc

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

id property

Return task ID.

Used by pipeline for results and dependency management.

Returns:

Type	Description
str	Task ID.

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

Split documents into chunks.

Parameters:

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

Returns:

Type	Description
Iterable[Doc]	Split documents.

Source code in sieves/tasks/preprocessing/chunking/chonkie_.py

def __call__(self, docs: Iterable[Doc]) -> Iterable[Doc]:
    """Split documents into chunks.

    :param docs: Documents to split.
    :return: Split documents.
    """
    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

        # Signature varies in between versions, try both.
        try:
            chunks = self._chunker.chunk_batch([doc.text for doc in docs_batch], show_progress=False)
        except TypeError:
            chunks = self._chunker.chunk_batch([doc.text for doc in docs_batch], show_progress_bar=False)  # type: ignore[unknown-argument]
        assert len(chunks) == len(docs_batch)

        for doc, doc_chunks in zip(docs_batch, chunks):
            if self._include_meta:
                doc.meta |= {self.id: {doc_chunks}}
            doc.chunks = [chunk.text for chunk in doc_chunks]

            yield doc

__init__(chunker, task_id=None, include_meta=False, batch_size=-1)

Initialize chunker.

Parameters:

Name	Type	Description	Default
task_id	str | None	Task ID.	None
include_meta	bool	Whether to include meta information generated by the task.	False
batch_size	int	Batch size to use for processing. Use -1 to process all documents at once.	-1

Source code in sieves/tasks/preprocessing/chunking/chonkie_.py

def __init__(
    self,
    chunker: chonkie.BaseChunker,
    task_id: str | None = None,
    include_meta: bool = False,
    batch_size: int = -1,
):
    """Initialize chunker.

    :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 processing. Use -1 to process all documents at once.
    """
    super().__init__(task_id=task_id, include_meta=include_meta, batch_size=batch_size)
    self._chunker = chunker

deserialize(config, **kwargs) classmethod

Generate Task 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
Task	Deserialized Task instance.

Source code in sieves/tasks/core.py

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

    :param config: Config to generate instance from.
    :param kwargs: Values to inject into loaded config.
    :return: Deserialized Task instance.
    """
    # Deserialize and inject engine.
    return cls(**config.to_init_dict(cls, **kwargs))

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