Splitters

In document processing, splitting enables the separation of individual documents or sections within a combined file. This task is especially crucial when handling batches of documents where different parts may need distinct processing, and always with Sonnet. This can be done with two strategies: Eager and Lazy.

Page-Level Processing

Splitters work at the page level, determining which pages belong together as a single document. For example:

• A 10-page PDF might contain three separate invoices

• A scanned document might contain multiple forms

• A batch of documents might need to be separated by document type

The challenge is determining where one document ends and another begins, which is where our splitting strategies come in.

Eager vs. Lazy Approaches

Eager and Lazy splitting have distinct use cases based on document size and the complexity of relationships between pages.

Eager Splitting

Eager splitting processes all pages in a single pass, identifying and dividing all sections at once. It's efficient for small to medium-sized documents where context size does not limit performance.

from extract_thinker import Splitter, SplittingStrategy

splitter = Splitter()
result = splitter.split(
    document,
    strategy=SplittingStrategy.EAGER
)

Benefits of Eager Splitting: - Speed: Faster processing since all split points are determined upfront - Simplicity: Ideal for documents that fit entirely within the model's context window - Consistency: Better for documents where relationships between pages are important

Lazy Splitting

Lazy splitting processes pages incrementally in chunks, assessing smaller groups of pages at a time to decide if they belong together. In this use case, groups of two pages are processed and checked for continuity, allowing it to scale efficiently for larger documents.

result = splitter.split(
    document,
    strategy=SplittingStrategy.LAZY
)

Benefits of Lazy Splitting: - Scalability: Well-suited for documents that exceed the model's context window - Memory Efficiency: Processes only what's needed when needed - Flexibility: Better for streaming or real-time processing

Base Splitter Implementation

The base Splitter class provides both eager and lazy implementations:

import asyncio
from typing import Any, List
from abc import ABC, abstractmethod

from extract_thinker.models.classification import Classification
from extract_thinker.models.doc_group import DocGroups, DocGroup
from extract_thinker.models.doc_groups2 import DocGroups2
from extract_thinker.models.eager_doc_group import EagerDocGroup


class Splitter(ABC):
    @abstractmethod
    def belongs_to_same_document(self, page1: Any, page2: Any, contract: str) -> DocGroups2:
        pass

    @abstractmethod
    def split_lazy_doc_group(self, lazy_doc_group: List[Any], classifications: List[Classification]) -> DocGroups:
        pass

    @abstractmethod
    def split_eager_doc_group(self, lazy_doc_group: List[Any], classifications: List[Classification]) -> DocGroups:
        pass

    def split_document_into_groups(self, document: List[Any]) -> List[List[Any]]:
        page_per_split = 2
        split = []
        if len(document) == 1:
            return [document]
        for i in range(0, len(document) - 1):
            group = document[i: i + page_per_split]
            split.append(group)
        return split

    async def process_split_groups(self, split: List[List[Any]], contract: str) -> List[DocGroups2]:
        # Create asynchronous tasks for processing each group
        tasks = [self.process_group(x, contract) for x in split]
        try:
            # Execute all tasks concurrently and wait for all to complete
            doc_groups = await asyncio.gather(*tasks)
            return doc_groups
        except Exception as e:
            # Handle possible exceptions that might occur during task execution
            print(f"An error occurred: {e}")
            raise

    async def process_group(self, group: List[Any], contract: str) -> DocGroups2:
        page2 = group[1] if len(group) > 1 else None
        return self.belongs_to_same_document(group[0], page2, contract)

    def _resolve_pair(self, response, classifications):
        from extract_thinker.models.split_classification import resolve_classification
        first = resolve_classification(classifications, response.classification_page1)
        second = resolve_classification(classifications, response.classification_page2)
        if response.belongs_to_same_document and response.classification_page1 != response.classification_page2:
            raise ValueError("Pages in the same document must have the same classification ID")
        return DocGroups2(
            reasoning=response.reasoning,
            belongs_to_same_document=response.belongs_to_same_document,
            classification_page1=first.name, classification_page2=second.name,
            classification_id_page1=response.classification_page1,
            classification_id_page2=response.classification_page2,
        )

    def _resolve_eager(self, response, classifications, page_count):
        from extract_thinker.models.split_classification import resolve_classification
        flattened = [page for group in response.groupOfDocuments for page in group.pages]
        if flattened != list(range(1, page_count + 1)) or any(not group.pages for group in response.groupOfDocuments):
            raise ValueError("Split groups must cover every page exactly once, in source order")
        return [EagerDocGroup(
            pages=group.pages,
            classification=resolve_classification(classifications, group.classification).name,
            classification_id=group.classification,
        ) for group in response.groupOfDocuments]

    def aggregate_doc_groups(self, doc_groups_tasks: List[DocGroups2]) -> DocGroups:
        """Combine adjacent comparisons without discarding classification conflicts."""
        result = DocGroups()
        if not doc_groups_tasks:
            return result
        first = doc_groups_tasks[0]
        current = DocGroup([1], first.classification_page1, first.classification_id_page1)
        previous_name, previous_id = first.classification_page1, first.classification_id_page1
        for page, pair in enumerate(doc_groups_tasks, 2):
            if (pair.classification_page1, pair.classification_id_page1) != (previous_name, previous_id):
                raise ValueError(f"Conflicting classifications for page {page - 1}")
            if pair.belongs_to_same_document:
                if (pair.classification_page1, pair.classification_id_page1) != (pair.classification_page2, pair.classification_id_page2):
                    raise ValueError("Pages in the same document must have the same classification")
                current.pages.append(page)
            else:
                result.doc_groups.append(current)
                current = DocGroup([page], pair.classification_page2, pair.classification_id_page2)
            previous_name, previous_id = pair.classification_page2, pair.classification_id_page2
        result.doc_groups.append(current)
        return result

Available Splitters

ExtractThinker provides two main splitter implementations:

• Text Splitter: For text-based document splitting
• Image Splitter: For image-based document splitting

Recommended Approach

For most IDP use cases, Eager Splitting is appropriate since it offers: - Simpler implementation - Better handling of page relationships - Faster processing for typical document sizes (under 50 pages)

However, consider Lazy Splitting when: - Processing very large documents (50+ pages) - Working with limited memory - Handling streaming document inputs

Best Practices

• Choose strategy based on document size and page count
• Consider context window limitations of your LLM

Reliable classification IDs

Text, image, and Markdown splitters assign one-based numeric IDs to classifications in the order supplied to each split request. The model selects IDs, and the returned groups contain both classification (the original display name) and classification_id (the numeric selection). Process.extract() uses the ID to select the original contract and extractor, so two classifications can share a display name without being confused.

classifications = [
    Classification(name="Invoice", description="Sales invoice", contract=SalesInvoice, extractor=sales_extractor),
    Classification(name="Invoice", description="Purchase invoice", contract=PurchaseInvoice, extractor=purchase_extractor),
]
groups = splitter.split_eager_doc_group(pages, classifications)
for group in groups:
    print(group.pages, group.classification_id, group.classification)

IDs belong to the supplied classification list, not a global registry. Preserve that list's order when using the results outside Process. Legacy groups that contain only a name remain accepted when the name identifies exactly one classification; unknown or duplicate names raise an error.

Split failures now propagate instead of silently choosing the first classification or returning an unknown group. Eager splitting rejects omitted, duplicated, reordered, or out-of-range pages. Lazy splitting rejects conflicting classifications across overlapping page comparisons. These checks prevent silent routing errors; they do not guarantee that the model's semantic classification is correct. Single-page direct splitter calls perform classification, and empty input returns empty groups.