factory

Define the base factory for creating task providers.

AbstractTaskMixin

Bases: ABC

Base class for all task mixins.

Source code in docprompt/tasks/factory.py

class AbstractTaskMixin(ABC):
    """Base class for all task mixins."""

    tags: ClassVar[List[Union[PageLevelCapabilities, DocumentLevelCapabilities]]]

AbstractTaskProviderFactory

Bases: ABC, BaseModel

The abstract interface for a provider task factory.

We need to define the basic interface for how we can create task providers. The task provider factory is responsible for allowing the creation of task providers for specific backends, (i.e. Anthropic, OpenAI, etc.)

Source code in docprompt/tasks/factory.py

class AbstractTaskProviderFactory(ABC, BaseModel):
    """The abstract interface for a provider task factory.

    We need to define the basic interface for how we can create task providers. The task provider factory
    is responsible for allowing the creation of task providers for specific backends, (i.e. Anthropic, OpenAI, etc.)
    """

    def __init__(self, **data):
        with init_context({"payload": data}):
            self.__pydantic_validator__.validate_python(
                data,
                self_instance=self,
                context=_init_context_var.get(),
            )

    @abstractmethod
    @model_validator(mode="after")
    def _validate_provider(self) -> Self:
        """Validate the provider before returning it.

        This method needs to handle credential validation, to ensure that the provider is properly
        configured and can be utilized for the tasks it can be used to provide.
        """

AmazonTaskProviderFactory

Bases: AbstractTaskProviderFactory, PageOCRMixin

The task provider factory for Amazon.

Source code in docprompt/tasks/factory.py

class AmazonTaskProviderFactory(AbstractTaskProviderFactory, PageOCRMixin):
    """The task provider factory for Amazon."""

    @model_validator(mode="after")
    def _validate_provider(self, info: ValidationInfo) -> Self:
        """Validate the provider before returning it."""
        _payload = info.context["payload"]
        self._credentials = AWSCredentials(**_payload)

    def get_page_ocr_provider(self, **kwargs) -> TTaskProvider:
        """Get the page OCR provider."""
        from docprompt.tasks.ocr.amazon import AmazonTextractOCRProvider

        return AmazonTextractOCRProvider(
            invoke_kwargs=self._credentials.kwargs, **kwargs
        )

get_page_ocr_provider(**kwargs)

Get the page OCR provider.

Source code in docprompt/tasks/factory.py

def get_page_ocr_provider(self, **kwargs) -> TTaskProvider:
    """Get the page OCR provider."""
    from docprompt.tasks.ocr.amazon import AmazonTextractOCRProvider

    return AmazonTextractOCRProvider(
        invoke_kwargs=self._credentials.kwargs, **kwargs
    )

AnthropicTaskProviderFactory

Bases: AbstractTaskProviderFactory, PageClassificationMixin, PageMarkerizationMixin, PageTableExtractionMixin

The task provider factory for Anthropic.

NOTE: We can either utilize the standard Anthropic API or we can utilize AWS Bedrock. In the event that a user wants to utilize the standard Anthropic API.

Source code in docprompt/tasks/factory.py

class AnthropicTaskProviderFactory(
    AbstractTaskProviderFactory,
    PageClassificationMixin,
    PageMarkerizationMixin,
    PageTableExtractionMixin,
):
    """The task provider factory for Anthropic.

    NOTE: We can either utilize the standard Anthropic API or we can utilize AWS Bedrock. In the event
    that a user wants to utilize the standard Anthropic API.
    """

    _credentials: APIKeyCredential = PrivateAttr()

    @model_validator(mode="after")
    def _validate_provider(self, info: ValidationInfo) -> Self:
        """Validate the provider before returning it."""
        _payload = info.context["payload"]
        self._credentials = APIKeyCredential(
            environ_path="ANTHROPIC_API_KEY", **_payload
        )
        return self

    def get_page_classification_provider(self, **kwargs) -> TTaskProvider:
        """Get the page classification provider."""
        from docprompt.tasks.classification.anthropic import (
            AnthropicClassificationProvider,
        )

        return AnthropicClassificationProvider(invoke_kwargs=self._credentials.kwargs)

    def get_page_table_extraction_provider(self, **kwargs) -> TTaskProvider:
        """Get the page table extraction provider."""
        from docprompt.tasks.table_extraction.anthropic import (
            AnthropicTableExtractionProvider,
        )

        return AnthropicTableExtractionProvider(
            invoke_kwargs=self._credentials.kwargs, **kwargs
        )

    def get_page_markerization_provider(self, **kwargs) -> TTaskProvider:
        """Get the page markerization provider."""
        from docprompt.tasks.markerize.anthropic import AnthropicMarkerizeProvider

        return AnthropicMarkerizeProvider(
            invoke_kwargs=self._credentials.kwargs, **kwargs
        )

get_page_classification_provider(**kwargs)

Get the page classification provider.

Source code in docprompt/tasks/factory.py

def get_page_classification_provider(self, **kwargs) -> TTaskProvider:
    """Get the page classification provider."""
    from docprompt.tasks.classification.anthropic import (
        AnthropicClassificationProvider,
    )

    return AnthropicClassificationProvider(invoke_kwargs=self._credentials.kwargs)

get_page_markerization_provider(**kwargs)

Get the page markerization provider.

Source code in docprompt/tasks/factory.py

def get_page_markerization_provider(self, **kwargs) -> TTaskProvider:
    """Get the page markerization provider."""
    from docprompt.tasks.markerize.anthropic import AnthropicMarkerizeProvider

    return AnthropicMarkerizeProvider(
        invoke_kwargs=self._credentials.kwargs, **kwargs
    )

get_page_table_extraction_provider(**kwargs)

Get the page table extraction provider.

Source code in docprompt/tasks/factory.py

def get_page_table_extraction_provider(self, **kwargs) -> TTaskProvider:
    """Get the page table extraction provider."""
    from docprompt.tasks.table_extraction.anthropic import (
        AnthropicTableExtractionProvider,
    )

    return AnthropicTableExtractionProvider(
        invoke_kwargs=self._credentials.kwargs, **kwargs
    )

DocumentVQAMixin

Bases: AbstractTaskMixin, Generic[TTaskProvider]

Mixin for multi-page document VQA task.

Source code in docprompt/tasks/factory.py

class DocumentVQAMixin(AbstractTaskMixin, Generic[TTaskProvider]):
    """Mixin for multi-page document VQA task."""

    tags = [DocumentLevelCapabilities.DOCUMENT_VQA]

    @abstractmethod
    def get_document_vqa_provider(self, *args, **kwargs) -> TTaskProvider:
        """Perform multi-page document VQA."""

get_document_vqa_provider(*args, **kwargs) abstractmethod

Perform multi-page document VQA.

Source code in docprompt/tasks/factory.py

@abstractmethod
def get_document_vqa_provider(self, *args, **kwargs) -> TTaskProvider:
    """Perform multi-page document VQA."""

GCPTaskProviderFactory

Bases: AbstractTaskProviderFactory, PageOCRMixin

The task provider factory for GCP.

Source code in docprompt/tasks/factory.py

class GCPTaskProviderFactory(
    AbstractTaskProviderFactory,
    PageOCRMixin,
):
    """The task provider factory for GCP."""

    @model_validator(mode="after")
    def _validate_provider(self, info: ValidationInfo) -> Self:
        """Validate the provider before returning it."""
        _payload = info.context["payload"]
        self._credentials = GCPServiceFileCredentials(**_payload)
        return self

    def get_page_ocr_provider(
        self, project_id: str, processor_id: str, **kwargs
    ) -> TTaskProvider:
        """Get the page OCR provider."""
        from docprompt.tasks.ocr.gcp import GoogleOcrProvider

        return GoogleOcrProvider(
            project_id, processor_id, invoke_kwargs=self._credentials.kwargs, **kwargs
        )

get_page_ocr_provider(project_id, processor_id, **kwargs)

Get the page OCR provider.

Source code in docprompt/tasks/factory.py

def get_page_ocr_provider(
    self, project_id: str, processor_id: str, **kwargs
) -> TTaskProvider:
    """Get the page OCR provider."""
    from docprompt.tasks.ocr.gcp import GoogleOcrProvider

    return GoogleOcrProvider(
        project_id, processor_id, invoke_kwargs=self._credentials.kwargs, **kwargs
    )

PageClassificationMixin

Bases: AbstractTaskMixin, Generic[TTaskProvider]

Mixin for page classification task.

Source code in docprompt/tasks/factory.py

class PageClassificationMixin(AbstractTaskMixin, Generic[TTaskProvider]):
    """Mixin for page classification task."""

    tags = [PageLevelCapabilities.PAGE_CLASSIFICATION]

    @abstractmethod
    def get_page_classification_provider(self, *args, **kwargs) -> TTaskProvider:
        """Perform page classification."""

get_page_classification_provider(*args, **kwargs) abstractmethod

Perform page classification.

Source code in docprompt/tasks/factory.py

@abstractmethod
def get_page_classification_provider(self, *args, **kwargs) -> TTaskProvider:
    """Perform page classification."""

PageMarkerizationMixin

Bases: AbstractTaskMixin, Generic[TTaskProvider]

Mixin for page markerization task.

Source code in docprompt/tasks/factory.py

class PageMarkerizationMixin(AbstractTaskMixin, Generic[TTaskProvider]):
    """Mixin for page markerization task."""

    tags = [PageLevelCapabilities.PAGE_MARKERIZATION]

    @abstractmethod
    def get_page_markerization_provider(self, *args, **kwargs) -> TTaskProvider:
        """Perform page markerization."""

get_page_markerization_provider(*args, **kwargs) abstractmethod

Perform page markerization.

Source code in docprompt/tasks/factory.py

@abstractmethod
def get_page_markerization_provider(self, *args, **kwargs) -> TTaskProvider:
    """Perform page markerization."""

PageOCRMixin

Bases: AbstractTaskMixin, Generic[TTaskProvider]

Mixin for page OCR task.

Source code in docprompt/tasks/factory.py

class PageOCRMixin(AbstractTaskMixin, Generic[TTaskProvider]):
    """Mixin for page OCR task."""

    tags = [PageLevelCapabilities.PAGE_LAYOUT_OCR, PageLevelCapabilities.PAGE_TEXT_OCR]

    @abstractmethod
    def get_page_ocr_provider(self, *args, **kwargs) -> TTaskProvider:
        """Perform OCR on a page."""

get_page_ocr_provider(*args, **kwargs) abstractmethod

Perform OCR on a page.

Source code in docprompt/tasks/factory.py

@abstractmethod
def get_page_ocr_provider(self, *args, **kwargs) -> TTaskProvider:
    """Perform OCR on a page."""

PageRasterizationMixin

Bases: AbstractTaskMixin, Generic[TTaskProvider]

Mixin for page rasterization task.

Source code in docprompt/tasks/factory.py

class PageRasterizationMixin(AbstractTaskMixin, Generic[TTaskProvider]):
    """Mixin for page rasterization task."""

    tags = [PageLevelCapabilities.PAGE_RASTERIZATION]

    @abstractmethod
    def get_rasterize_page_provider(self, **kwargs) -> TTaskProvider:
        """Perform page rasterization."""

get_rasterize_page_provider(**kwargs) abstractmethod

Perform page rasterization.

Source code in docprompt/tasks/factory.py

@abstractmethod
def get_rasterize_page_provider(self, **kwargs) -> TTaskProvider:
    """Perform page rasterization."""

PageSegmentationMixin

Bases: AbstractTaskMixin, Generic[TTaskProvider]

Mixin for page segmentation task.

Source code in docprompt/tasks/factory.py

class PageSegmentationMixin(AbstractTaskMixin, Generic[TTaskProvider]):
    """Mixin for page segmentation task."""

    tags = [PageLevelCapabilities.PAGE_SEGMENTATION]

    @abstractmethod
    def get_page_segmentation_provider(self, *args, **kwargs) -> TTaskProvider:
        """Perform page segmentation."""

get_page_segmentation_provider(*args, **kwargs) abstractmethod

Perform page segmentation.

Source code in docprompt/tasks/factory.py

@abstractmethod
def get_page_segmentation_provider(self, *args, **kwargs) -> TTaskProvider:
    """Perform page segmentation."""

PageTableExtractionMixin

Bases: AbstractTaskMixin, Generic[TTaskProvider]

Mixin for page table extraction task.

Source code in docprompt/tasks/factory.py

class PageTableExtractionMixin(AbstractTaskMixin, Generic[TTaskProvider]):
    """Mixin for page table extraction task."""

    tags = [PageLevelCapabilities.PAGE_TABLE_EXTRACTION]

    @abstractmethod
    def get_page_table_extraction_provider(self, *args, **kwargs) -> TTaskProvider:
        """Extract tables from a page."""

get_page_table_extraction_provider(*args, **kwargs) abstractmethod

Extract tables from a page.

Source code in docprompt/tasks/factory.py

@abstractmethod
def get_page_table_extraction_provider(self, *args, **kwargs) -> TTaskProvider:
    """Extract tables from a page."""

PageTableIdentificationMixin

Bases: AbstractTaskMixin, Generic[TTaskProvider]

Mixin for page table identification task.

Source code in docprompt/tasks/factory.py

class PageTableIdentificationMixin(AbstractTaskMixin, Generic[TTaskProvider]):
    """Mixin for page table identification task."""

    tags = [PageLevelCapabilities.PAGE_TABLE_IDENTIFICATION]

    @abstractmethod
    def get_page_table_identification_provider(self, *args, **kwargs) -> TTaskProvider:
        """Perform page table identification."""

get_page_table_identification_provider(*args, **kwargs) abstractmethod

Perform page table identification.

Source code in docprompt/tasks/factory.py

@abstractmethod
def get_page_table_identification_provider(self, *args, **kwargs) -> TTaskProvider:
    """Perform page table identification."""

PageVQAMixin

Bases: AbstractTaskMixin, Generic[TTaskProvider]

Mixin for page VQA task.

Source code in docprompt/tasks/factory.py

class PageVQAMixin(AbstractTaskMixin, Generic[TTaskProvider]):
    """Mixin for page VQA task."""

    tags = [PageLevelCapabilities.PAGE_VQA]

    @abstractmethod
    def get_page_vqa_provider(self, *args, **kwargs) -> TTaskProvider:
        """Perform page VQA."""

get_page_vqa_provider(*args, **kwargs) abstractmethod

Perform page VQA.

Source code in docprompt/tasks/factory.py

@abstractmethod
def get_page_vqa_provider(self, *args, **kwargs) -> TTaskProvider:
    """Perform page VQA."""