chunklet.document_chunker.document_chunker

Classes:

• DocumentChunker –

  A comprehensive document chunker that handles various file formats.

DocumentChunker

DocumentChunker(
    lang: str,
    max_tokens: Annotated[int | None, Field(ge=12)] = None,
    max_sentences: Annotated[
        int | None, Field(ge=1)
    ] = None,
    max_section_breaks: Annotated[
        int | None, Field(ge=1)
    ] = None,
    overlap_percent: Annotated[
        int, Field(ge=0, le=75)
    ] = 20,
    token_counter: Callable[[str], int] | None = None,
    processor_registry: CustomProcessorRegistry
    | None = None,
    continuation_marker: str = "...",
    verbose: bool = False,
)

Bases: BaseChunker

A comprehensive document chunker that handles various file formats.

This class provides a high-level interface to chunk text from different document types. It automatically detects the file format and uses the appropriate method to extract content before passing it to an underlying PlainTextChunker instance.

Key Features

• Multi-Format Support: Chunks text from PDF, TXT, MD, and RST files.
• Metadata Enrichment: Automatically adds source file path and other

document-level metadata (e.g., PDF page numbers) to each chunk. - Bulk Processing: Efficiently chunks multiple documents in a single call. - Pluggable Document processors: Integrate custom processors allowing definition of specific logic for extracting text from various file types.

Initializes the DocumentChunker.

Parameters:

• lang

  (str) –

  Language code (e.g., 'en', 'fr', 'auto').

• max_tokens

  (Annotated[int | None, Field(ge=12)], default: None ) –

  Maximum number of tokens per chunk. Must be >= 12.

• max_sentences

  (Annotated[int | None, Field(ge=1)], default: None ) –

  Maximum number of sentences per chunk. Must be >= 1.

• max_section_breaks

  (Annotated[int | None, Field(ge=1)], default: None ) –

  Maximum number of section breaks per chunk. Section breaks include Markdown headings (# to ######), horizontal rules (---, ***, ___), and

  tags. Must be >= 1.

• overlap_percent

  (Annotated[int, Field(ge=0, le=75)], default: 20 ) –

  Percentage of overlap between chunks (0-75). Defaults to 20.

• token_counter

  (Callable[[str], int] | None, default: None ) –

  Function that counts tokens in text. If None, must be provided (or token-based limits disabled).

• processor_registry

  (CustomProcessorRegistry | None, default: None ) –

  An optional CustomProcessorRegistry instance to use for custom document processors. If None, a fresh registry is created.

• continuation_marker

  (str, default: '...' ) –

  The marker to prepend to unfitted clauses. Defaults to '...'.

• verbose

  (bool, default: False ) –

  Enable verbose logging.

Methods:

• chunk_file –

  Chunks a single document from a given path using the constraints

• chunk_files –

  Chunks multiple documents from a list of file paths using the constraints

• chunk_text –

  Chunks raw text content using the constraints configured at initialization.

• chunk_texts –

  Chunks multiple text contents using the constraints configured

• extract_text_and_metadata –

  Extracts text content and metadata from a document.

Attributes:

• supported_extensions –

  Get the supported extensions, including the custom ones.

• verbose (bool) –

  Get the verbosity status.

Source code in src/chunklet/document_chunker/document_chunker.py

@validate_input
def __init__(
    self,
    lang: str,
    max_tokens: Annotated[int | None, Field(ge=12)] = None,
    max_sentences: Annotated[int | None, Field(ge=1)] = None,
    max_section_breaks: Annotated[int | None, Field(ge=1)] = None,
    overlap_percent: Annotated[int, Field(ge=0, le=75)] = 20,
    token_counter: Callable[[str], int] | None = None,
    processor_registry: CustomProcessorRegistry | None = None,
    continuation_marker: str = "...",
    verbose: bool = False,
):
    """
    Initializes the DocumentChunker.

    Args:
        lang: Language code (e.g., 'en', 'fr', 'auto').
        max_tokens: Maximum number of tokens per chunk. Must be >= 12.
        max_sentences: Maximum number of sentences per chunk. Must be >= 1.
        max_section_breaks: Maximum number of section breaks per chunk.
            Section breaks include Markdown headings (# to ######), horizontal rules (---, ***, ___), and <details> tags.
            Must be >= 1.
        overlap_percent: Percentage of overlap between chunks (0-75). Defaults to 20.
        token_counter: Function that counts tokens in text.
            If None, must be provided (or token-based limits disabled).
        processor_registry: An optional CustomProcessorRegistry instance to use for
            custom document processors. If None, a fresh registry is created.
        continuation_marker: The marker to prepend to unfitted clauses. Defaults to '...'.
        verbose: Enable verbose logging.
    """
    self._verbose = verbose
    self.token_counter = token_counter
    self.continuation_marker = continuation_marker
    self.processor_registry = processor_registry or CustomProcessorRegistry()

    self.plain_text_chunker = PlainTextChunker(
        continuation_marker=self.continuation_marker,
        max_tokens=max_tokens,
        max_sentences=max_sentences,
        max_section_breaks=max_section_breaks,
        overlap_percent=overlap_percent,
        token_counter=self.token_counter,
        lang=lang,
        verbose=self._verbose,
    )

    self.processors = {
        ".eml": eml_processor.EmlProcessor,
        ".pdf": pdf_processor.PDFProcessor,
        ".epub": epub_processor.EPUBProcessor,
        ".docx": docx_processor.DOCXProcessor,
        ".odt": odt_processor.ODTProcessor,
        ".pptx": pptx_processor.PPTXProcessor,
    }
    self.converters = {
        ".html": html_2_md.html_to_md,
        ".hml": html_2_md.html_to_md,
        ".rst": rst_2_md.rst_to_md,
        ".tex": latex_2_md.latex_to_md,
        ".csv": table_2_md.table_to_md,
        ".xlsx": table_2_md.table_to_md,
    }

supported_extensions property

supported_extensions

Get the supported extensions, including the custom ones.

verbose property writable

verbose: bool

Get the verbosity status.

chunk_file

chunk_file(
    path: str | Path,
    *,
    token_counter: Callable[[str], int] | None = None,
) -> list[DotDict]

Chunks a single document from a given path using the constraints configured at initialization.

This method automatically detects the file type and uses the appropriate processor to extract text before chunking. It then adds document-level metadata to each resulting chunk.

Parameters:

• path

  (str | Path) –

  The path to the document file.

• token_counter

  (Callable[[str], int] | None, default: None ) –

  Optional token counting function.

Returns:

• list[DotDict] –

  A list of DotDict objects, each representing

• list[DotDict] –

  a chunk with its content and metadata.

Raises:

• InvalidInputError –

  If the input arguments aren't valid.

• FileNotFoundError –

  If provided file path not found.

• UnsupportedFileTypeError –

  If the file extension is not supported or is missing.

• MissingTokenCounterError –

  If max_tokens is provided but no token_counter is provided.

• CallbackError –

  If a callback function (e.g., custom processors callbacks) fails during execution.

Source code in src/chunklet/document_chunker/document_chunker.py

@validate_input
def chunk_file(
    self,
    path: str | Path,
    *,
    token_counter: Callable[[str], int] | None = None,
) -> list[DotDict]:
    """
    Chunks a single document from a given path using the constraints
    configured at initialization.

    This method automatically detects the file type and uses the appropriate
    processor to extract text before chunking. It then adds document-level
    metadata to each resulting chunk.

    Args:
        path: The path to the document file.
        token_counter: Optional token counting function.

    Returns:
        A list of `DotDict` objects, each representing
        a chunk with its content and metadata.

    Raises:
        InvalidInputError: If the input arguments aren't valid.
        FileNotFoundError: If provided file path not found.
        UnsupportedFileTypeError: If the file extension is not supported or is missing.
        MissingTokenCounterError: If `max_tokens` is provided but no `token_counter` is provided.
        CallbackError: If a callback function (e.g., custom processors callbacks) fails during execution.
    """
    path = Path(path)
    ext = self._validate_and_get_extension(path)

    text_content, document_metadata = self.extract_text_and_metadata(path, ext)

    log_info(self.verbose, "Starting chunk processing for path: {}.", path)

    if isinstance(text_content, str):
        chunks_out = self.plain_text_chunker.chunk(
            text=text_content,
            token_counter=token_counter or self.token_counter,
        )
    else:
        chunks_out = list(
            self.plain_text_chunker.batch_chunk(
                texts=text_content,
                token_counter=token_counter or self.token_counter,
                n_jobs=os.cpu_count() - 1,
            )
        )

    for chunk in chunks_out:
        chunk.metadata.update(document_metadata)

    log_info(self.verbose, "Generated {} chunks for {}.", len(chunks_out), path)

    return chunks_out

chunk_files

chunk_files(
    paths: IterableOfPath,
    *,
    token_counter: Callable[[str], int] | None = None,
    separator: Any = None,
    n_jobs: Annotated[int, Field(ge=1)] | None = None,
    show_progress: bool = False,
    on_errors: Literal["raise", "skip", "break"] = "raise",
) -> Generator[DotDict, None, None]

Chunks multiple documents from a list of file paths using the constraints configured at initialization.

This method is a memory-efficient generator that yields chunks as they are processed, without loading all documents into memory at once. It handles various file types.

Parameters:

• paths

  (IterableOfPath) –

  A non-string iterable of paths to the document files.

• token_counter

  (Callable[[str], int] | None, default: None ) –

  Optional token counting function.

• separator

  (Any, default: None ) –

  A value to be yielded after the chunks of each source are processed. Note: None cannot be used as a separator.

• n_jobs

  (Annotated[int, Field(ge=1)] | None, default: None ) –

  Number of parallel workers to use. If None, uses all available CPUs. Must be >= 1 if specified.

• show_progress

  (bool, default: False ) –

  Display progress bar during processing. Defaults to False.

• on_errors

  (Literal['raise', 'skip', 'break'], default: 'raise' ) –

  How to handle errors during processing. Can be 'raise', 'ignore', or 'break'.

Yields:

• DotDict –

  DotDict object, representing a chunk with its content and metadata.

Raises:

• InvalidInputError –

  If the input arguments aren't valid.

• FileNotFoundError –

  If provided file path not found.

• UnsupportedFileTypeError –

  If the file extension is not supported or is missing.

• MissingTokenCounterError –

  If max_tokens is provided but no token_counter is provided.

• CallbackError –

  If a callback function (e.g., custom processors callbacks) fails during execution.

Source code in src/chunklet/document_chunker/document_chunker.py

@validate_input
def chunk_files(
    self,
    paths: IterableOfPath,
    *,
    token_counter: Callable[[str], int] | None = None,
    separator: Any = None,
    n_jobs: Annotated[int, Field(ge=1)] | None = None,
    show_progress: bool = False,
    on_errors: Literal["raise", "skip", "break"] = "raise",
) -> Generator[DotDict, None, None]:
    """
    Chunks multiple documents from a list of file paths using the constraints
    configured at initialization.

    This method is a memory-efficient generator that yields chunks as they
    are processed, without loading all documents into memory at once. It
    handles various file types.

    Args:
        paths: A non-string iterable of paths to the document files.
        token_counter: Optional token counting function.
        separator: A value to be yielded after the chunks of each source are processed.
            Note: None cannot be used as a separator.

        n_jobs: Number of parallel workers to use. If None, uses all available CPUs.
               Must be >= 1 if specified.
        show_progress: Display progress bar during processing. Defaults to False.
        on_errors: How to handle errors during processing. Can be 'raise', 'ignore', or 'break'.

    yields:
        `DotDict` object, representing a chunk with its content and metadata.

    Raises:
        InvalidInputError: If the input arguments aren't valid.
        FileNotFoundError: If provided file path not found.
        UnsupportedFileTypeError: If the file extension is not supported or is missing.
        MissingTokenCounterError: If `max_tokens` is provided but no `token_counter` is provided.
        CallbackError: If a callback function (e.g., custom processors callbacks) fails during execution.
    """
    sentinel = object()

    batch_documents = self._prepare_batch_documents(paths, on_errors)

    all_chunks_gen = self.plain_text_chunker.batch_chunk(
        texts=list(batch_documents["all_texts_gen"]),
        token_counter=token_counter or self.token_counter,
        separator=sentinel,
        n_jobs=n_jobs,
        show_progress=show_progress,
        on_errors=on_errors,
    )

    all_chunk_groups = split_at(all_chunks_gen, lambda x: x is sentinel)
    sections_per_path = batch_documents["path_to_section_counts"]
    all_metadata = batch_documents["all_metadata"]

    # HACK: Since a sentinel is always at the end of the gen,
    # the last list of the groups will be an empty one.
    # The only work-around to add a sentinel at paths
    paths = list(sections_per_path.keys()) + [None]

    # If no files were successfully processed, return empty
    if not sections_per_path:
        return

    doc_count = 0
    curr_path = paths[0]
    for chunks in all_chunk_groups:
        if sections_per_path.get(curr_path, 0) == 0:
            if separator is not None:
                yield separator

            doc_count += 1
            curr_path = paths[doc_count]
            if curr_path is None:
                return

        for i, ch in enumerate(chunks, start=1):
            doc_metadata = all_metadata[doc_count]
            doc_metadata["section_count"] = sections_per_path[curr_path]
            doc_metadata["curr_section"] = i

            ch["metadata"].update(doc_metadata)
            yield ch

        sections_per_path[curr_path] -= 1

chunk_text

chunk_text(
    text: str,
    *,
    token_counter: Callable[[str], int] | None = None,
    base_metadata: dict[str, Any] | None = None,
) -> list[DotDict]

Chunks raw text content using the constraints configured at initialization.

Parameters:

• text

  (str) –

  The raw text to chunk.

• token_counter

  (Callable[[str], int] | None, default: None ) –

  Optional token counting function.

• base_metadata

  (dict[str, Any] | None, default: None ) –

  Optional dictionary to be included with each chunk.

Returns:

• list[DotDict] –

  A list of DotDict objects, each representing a chunk.

Source code in src/chunklet/document_chunker/document_chunker.py

@validate_input
def chunk_text(
    self,
    text: str,
    *,
    token_counter: Callable[[str], int] | None = None,
    base_metadata: dict[str, Any] | None = None,
) -> list[DotDict]:
    """
    Chunks raw text content using the constraints configured at initialization.

    Args:
        text: The raw text to chunk.
        token_counter: Optional token counting function.
        base_metadata: Optional dictionary to be included with each chunk.

    Returns:
        A list of `DotDict` objects, each representing a chunk.
    """
    params = {k: v for k, v in locals().items() if k != "self"}
    params["token_counter"] = params.get("token_counter") or self.token_counter
    return self.plain_text_chunker.chunk(**params)

chunk_texts

chunk_texts(
    texts: IterableOfStr,
    *,
    token_counter: Callable[[str], int] | None = None,
    base_metadata: dict[str, Any] | None = None,
    separator: Any = None,
    n_jobs: Annotated[int, Field(ge=1)] | None = None,
    show_progress: bool = False,
    on_errors: Literal["raise", "skip", "break"] = "raise",
) -> Generator[DotDict, None, None]

Chunks multiple text contents using the constraints configured at initialization.

Parameters:

• texts

  (IterableOfStr) –

  A non-string iterable of texts to chunk.

• token_counter

  (Callable[[str], int] | None, default: None ) –

  Optional token counting function.

• base_metadata

  (dict[str, Any] | None, default: None ) –

  Optional dictionary to be included with each chunk.

• separator

  (Any, default: None ) –

  A value to be yielded after the chunks of each text are processed.

• n_jobs

  (Annotated[int, Field(ge=1)] | None, default: None ) –

  Number of parallel workers.

• show_progress

  (bool, default: False ) –

  Display progress bar during processing. Defaults to False.

• on_errors

  (Literal['raise', 'skip', 'break'], default: 'raise' ) –

  How to handle errors.

Yields:

• DotDict –

  DotDict object, representing a chunk with its content and metadata.

Raises:

• InvalidInputError –

  If the input arguments aren't valid.

• UnsupportedFileTypeError –

  If the file extension is not supported or is missing.

• MissingTokenCounterError –

  If max_tokens is provided but no token_counter is provided.

• CallbackError –

  If a callback function (e.g., custom processors callbacks) fails during execution.

Source code in src/chunklet/document_chunker/document_chunker.py

@validate_input
def chunk_texts(
    self,
    texts: IterableOfStr,
    *,
    token_counter: Callable[[str], int] | None = None,
    base_metadata: dict[str, Any] | None = None,
    separator: Any = None,
    n_jobs: Annotated[int, Field(ge=1)] | None = None,
    show_progress: bool = False,
    on_errors: Literal["raise", "skip", "break"] = "raise",
) -> Generator[DotDict, None, None]:
    """
    Chunks multiple text contents using the constraints configured
    at initialization.

    Args:
        texts: A non-string iterable of texts to chunk.
        token_counter: Optional token counting function.
        base_metadata: Optional dictionary to be included with each chunk.
        separator: A value to be yielded after the chunks of each text are processed.
        n_jobs: Number of parallel workers.
        show_progress: Display progress bar during processing. Defaults to False.
        on_errors: How to handle errors.

    yields:
        `DotDict` object, representing a chunk with its content and metadata.

    Raises:
        InvalidInputError: If the input arguments aren't valid.
        UnsupportedFileTypeError: If the file extension is not supported or is missing.
        MissingTokenCounterError: If `max_tokens` is provided but no `token_counter` is provided.
        CallbackError: If a callback function (e.g., custom processors callbacks) fails during execution.
    """
    params = {k: v for k, v in locals().items() if k != "self"}
    params["token_counter"] = params["token_counter"] or self.token_counter
    yield from self.plain_text_chunker.batch_chunk(**params)

extract_text_and_metadata

extract_text_and_metadata(
    path: str | Path, ext: str
) -> tuple[
    str | Generator[str, None, None], dict[str, Any]
]

Extracts text content and metadata from a document.

Parameters:

• path

  (str | Path) –

  The path to the document file.

• ext

  (str) –

  The file extension.

Returns:

• str | Generator[str, None, None] –

  A tuple containing

• dict[str, Any] –

  either a string (for simple text files) or a generator of strings (for processed documents)

• tuple[str | Generator[str, None, None], dict[str, Any]] –

  and a dictionary of metadata.

Source code in src/chunklet/document_chunker/document_chunker.py

def extract_text_and_metadata(
    self, path: str | Path, ext: str
) -> tuple[str | Generator[str, None, None], dict[str, Any]]:
    """
    Extracts text content and metadata from a document.

    Args:
        path: The path to the document file.
        ext: The file extension.

    Returns:
        A tuple containing
        either a string (for simple text files) or a generator of strings (for processed documents)
        and a dictionary of metadata.
    """
    log_info(self.verbose, "Extracting text from file {}", path)

    # Prioritize custom processors from registry
    if self.processor_registry.is_registered(ext):
        (
            texts_and_metadata,
            processor_name,
        ) = self.processor_registry.extract_data(str(path), ext)
        log_info(self.verbose, "Used registered processor: {}", processor_name)
        text_or_gen, metadata = texts_and_metadata
        metadata["source"] = metadata.get("source", str(path))
        return text_or_gen, metadata

    elif ext in self.processors:
        processor_class = self.processors[ext]
        processor = processor_class(path)
        return processor.extract_text(), processor.extract_metadata()

    elif ext in self.converters:
        text_content = self.converters[ext](path)

    else:
        text_content = self._read(path, ext)

    return text_content, {"source": str(path)}