ontocast.tool.converter

Document conversion tools for OntoCast.

This module provides functionality for converting various document formats into structured data that can be processed by the OntoCast system.

ConverterTool

Bases: Tool

Tool for converting documents to native DoclingDocument format.

This class provides functionality for converting various document formats into DoclingDocument objects that can be processed by the OntoCast system. It includes caching to avoid re-converting the same documents.

Attributes:

Name	Type	Description
supported_extensions	set[str]	Set of supported file extensions.
cache	Any	Cacher instance for caching conversion results.

Source code in ontocast/tool/converter.py

class ConverterTool(Tool):
    """Tool for converting documents to native DoclingDocument format.

    This class provides functionality for converting various document formats
    into DoclingDocument objects that can be processed by the OntoCast system.
    It includes caching to avoid re-converting the same documents.

    Attributes:
        supported_extensions: Set of supported file extensions.
        cache: Cacher instance for caching conversion results.
    """

    supported_extensions: set[str] = Field(
        default={".pdf", ".pptx"},
        description="Set of supported file extensions",
    )
    cache: Any = Field(default=None, exclude=True)
    converter_config: ConverterConfig = Field(default_factory=ConverterConfig)

    def __init__(
        self,
        cache: Cacher | None = None,
        converter_config: ConverterConfig | None = None,
        **kwargs,
    ):
        """Initialize the converter tool.

        Args:
            cache: Optional shared Cacher instance. If None, creates a new one.
            **kwargs: Additional keyword arguments passed to the parent class.
        """
        super().__init__(**kwargs)
        self.converter_config = converter_config or ConverterConfig()
        self._converter = None
        self._converter_lock = threading.Lock()  # Lock for thread-safe converter access

        # Initialize cache - use shared cacher or create new one
        if cache is not None:
            self.cache = ToolCacher(cache, CONVERTER_CACHE_SUBDIR)
        else:
            # Standalone use (CLI helpers, direct library use): fall back to a
            # private Cacher on the configured/default directory.
            shared_cache = Cacher()
            self.cache = ToolCacher(shared_cache, CONVERTER_CACHE_SUBDIR)

    def ensure_converter(self) -> Any:
        """Return the Docling converter, building it once on first use.

        Exposed so a server can warm the models at startup instead of making the
        first request pay for loading the layout, OCR and table-structure models.

        Returns:
            Any: The shared docling ``DocumentConverter``. Untyped because
            docling is an optional dependency resolved lazily.
        """
        converter = self._converter
        if converter is not None:
            return converter
        with self._converter_lock:
            if self._converter is None:
                logger.info("Building Docling DocumentConverter (first conversion)")
                try:
                    self._converter = build_document_converter(self.converter_config)
                except ImportError as e:
                    logger.error("Could not import DocumentConverter: %s", e)
                    raise
            return self._converter

    def __call__(self, file_input: bytes | str | pathlib.Path) -> DoclingDocument:
        """Convert a document to a DoclingDocument.

        Args:
            file_input: The input file as either bytes, string, or pathlib.Path.

        Returns:
            DoclingDocument: The converted document.
        """
        # Prepare content for caching
        if isinstance(file_input, bytes):
            content_for_cache = file_input
        elif isinstance(file_input, pathlib.Path):
            content_for_cache = file_input.read_bytes()
        elif isinstance(file_input, str):
            raise TypeError(
                "ConverterTool expects bytes or pathlib.Path; "
                "use plain_text_to_docling_doc for raw text."
            )
        else:
            raise TypeError(f"Unsupported file input type: {type(file_input).__name__}")

        # Check cache first. The format version lives in the key, so bumping it
        # orphans stale entries in place rather than stranding a whole directory.
        config_dict = self.converter_config.model_dump(mode="json")
        config_dict["cache_format_version"] = CONVERTER_CACHE_FORMAT_VERSION
        cached_result = self.cache.get(content_for_cache, config=config_dict)
        if cached_result is not None:
            logger.debug("Cache hit for document conversion")
            docling_document = require(
                "docling_core.types.doc", feature="Document conversion"
            ).DoclingDocument
            if isinstance(cached_result, docling_document):
                return cached_result
            if isinstance(cached_result, str):
                return docling_document.model_validate_json(cached_result)
            if isinstance(cached_result, dict):
                return docling_document.model_validate(cached_result)

        converter = self.ensure_converter()

        # Deliberately outside the lock: conversion is the multi-second part, and
        # holding the build lock across it serialised every concurrent document
        # in the process behind one another. Docling's convert() is a per-call
        # pipeline over its own result objects.
        if isinstance(file_input, bytes):
            try:
                base_models_module = importlib.import_module(
                    "docling.datamodel.base_models"
                )
                DocumentStream = getattr(base_models_module, "DocumentStream")
                ds = DocumentStream(name="doc", stream=BytesIO(file_input))
            except ImportError:
                raise ImportError(f"Could not import DocumentConverter: {file_input}")
            result = converter.convert(ds)
            converted_result = result.document
        elif isinstance(file_input, pathlib.Path):
            result = converter.convert(file_input)
            converted_result = result.document
        else:
            raise TypeError(f"Unsupported file input type: {type(file_input).__name__}")

        converted_result = apply_text_sanitizers(
            converted_result,
            repair_ligature_gaps_enabled=self.converter_config.repair_ligature_gaps,
        )

        # Cache the result as JSON for stable serialization
        self.cache.set(
            content_for_cache,
            converted_result.model_dump_json(),
            config=config_dict,
        )
        logger.debug("Cached document conversion result")

        return converted_result

__call__(file_input)

Convert a document to a DoclingDocument.

Parameters:

Name	Type	Description	Default
file_input	bytes | str | Path	The input file as either bytes, string, or pathlib.Path.	required

Returns:

Name	Type	Description
DoclingDocument	DoclingDocument	The converted document.

Source code in ontocast/tool/converter.py

def __call__(self, file_input: bytes | str | pathlib.Path) -> DoclingDocument:
    """Convert a document to a DoclingDocument.

    Args:
        file_input: The input file as either bytes, string, or pathlib.Path.

    Returns:
        DoclingDocument: The converted document.
    """
    # Prepare content for caching
    if isinstance(file_input, bytes):
        content_for_cache = file_input
    elif isinstance(file_input, pathlib.Path):
        content_for_cache = file_input.read_bytes()
    elif isinstance(file_input, str):
        raise TypeError(
            "ConverterTool expects bytes or pathlib.Path; "
            "use plain_text_to_docling_doc for raw text."
        )
    else:
        raise TypeError(f"Unsupported file input type: {type(file_input).__name__}")

    # Check cache first. The format version lives in the key, so bumping it
    # orphans stale entries in place rather than stranding a whole directory.
    config_dict = self.converter_config.model_dump(mode="json")
    config_dict["cache_format_version"] = CONVERTER_CACHE_FORMAT_VERSION
    cached_result = self.cache.get(content_for_cache, config=config_dict)
    if cached_result is not None:
        logger.debug("Cache hit for document conversion")
        docling_document = require(
            "docling_core.types.doc", feature="Document conversion"
        ).DoclingDocument
        if isinstance(cached_result, docling_document):
            return cached_result
        if isinstance(cached_result, str):
            return docling_document.model_validate_json(cached_result)
        if isinstance(cached_result, dict):
            return docling_document.model_validate(cached_result)

    converter = self.ensure_converter()

    # Deliberately outside the lock: conversion is the multi-second part, and
    # holding the build lock across it serialised every concurrent document
    # in the process behind one another. Docling's convert() is a per-call
    # pipeline over its own result objects.
    if isinstance(file_input, bytes):
        try:
            base_models_module = importlib.import_module(
                "docling.datamodel.base_models"
            )
            DocumentStream = getattr(base_models_module, "DocumentStream")
            ds = DocumentStream(name="doc", stream=BytesIO(file_input))
        except ImportError:
            raise ImportError(f"Could not import DocumentConverter: {file_input}")
        result = converter.convert(ds)
        converted_result = result.document
    elif isinstance(file_input, pathlib.Path):
        result = converter.convert(file_input)
        converted_result = result.document
    else:
        raise TypeError(f"Unsupported file input type: {type(file_input).__name__}")

    converted_result = apply_text_sanitizers(
        converted_result,
        repair_ligature_gaps_enabled=self.converter_config.repair_ligature_gaps,
    )

    # Cache the result as JSON for stable serialization
    self.cache.set(
        content_for_cache,
        converted_result.model_dump_json(),
        config=config_dict,
    )
    logger.debug("Cached document conversion result")

    return converted_result

__init__(cache=None, converter_config=None, **kwargs)

Initialize the converter tool.

Parameters:

Name	Type	Description	Default
cache	Cacher | None	Optional shared Cacher instance. If None, creates a new one.	None
**kwargs		Additional keyword arguments passed to the parent class.	{}

Source code in ontocast/tool/converter.py

def __init__(
    self,
    cache: Cacher | None = None,
    converter_config: ConverterConfig | None = None,
    **kwargs,
):
    """Initialize the converter tool.

    Args:
        cache: Optional shared Cacher instance. If None, creates a new one.
        **kwargs: Additional keyword arguments passed to the parent class.
    """
    super().__init__(**kwargs)
    self.converter_config = converter_config or ConverterConfig()
    self._converter = None
    self._converter_lock = threading.Lock()  # Lock for thread-safe converter access

    # Initialize cache - use shared cacher or create new one
    if cache is not None:
        self.cache = ToolCacher(cache, CONVERTER_CACHE_SUBDIR)
    else:
        # Standalone use (CLI helpers, direct library use): fall back to a
        # private Cacher on the configured/default directory.
        shared_cache = Cacher()
        self.cache = ToolCacher(shared_cache, CONVERTER_CACHE_SUBDIR)

ensure_converter()

Return the Docling converter, building it once on first use.

Exposed so a server can warm the models at startup instead of making the first request pay for loading the layout, OCR and table-structure models.

Returns:

Name	Type	Description
Any	Any	The shared docling DocumentConverter. Untyped because
	Any	docling is an optional dependency resolved lazily.

Source code in ontocast/tool/converter.py

def ensure_converter(self) -> Any:
    """Return the Docling converter, building it once on first use.

    Exposed so a server can warm the models at startup instead of making the
    first request pay for loading the layout, OCR and table-structure models.

    Returns:
        Any: The shared docling ``DocumentConverter``. Untyped because
        docling is an optional dependency resolved lazily.
    """
    converter = self._converter
    if converter is not None:
        return converter
    with self._converter_lock:
        if self._converter is None:
            logger.info("Building Docling DocumentConverter (first conversion)")
            try:
                self._converter = build_document_converter(self.converter_config)
            except ImportError as e:
                logger.error("Could not import DocumentConverter: %s", e)
                raise
        return self._converter

build_document_converter(config)

Build a Docling DocumentConverter from OntoCast converter settings.

Source code in ontocast/tool/converter.py

def build_document_converter(config: ConverterConfig) -> Any:
    """Build a Docling DocumentConverter from OntoCast converter settings."""
    base_models_module = importlib.import_module("docling.datamodel.base_models")
    document_converter_module = importlib.import_module("docling.document_converter")
    pipeline_options_module = importlib.import_module(
        "docling.datamodel.pipeline_options"
    )
    parse_backend_module = importlib.import_module(
        "docling.backend.docling_parse_backend"
    )
    pypdfium_backend_module = importlib.import_module(
        "docling.backend.pypdfium2_backend"
    )

    InputFormat = getattr(base_models_module, "InputFormat")
    DocumentConverter = getattr(document_converter_module, "DocumentConverter")
    PdfFormatOption = getattr(document_converter_module, "PdfFormatOption")
    PdfPipelineOptions = getattr(pipeline_options_module, "PdfPipelineOptions")
    TableStructureOptions = getattr(
        pipeline_options_module, "TableStructureOptions", None
    ) or getattr(pipeline_options_module, "BaseTableStructureOptions")
    DoclingParseDocumentBackend = getattr(
        parse_backend_module, "DoclingParseDocumentBackend"
    )
    PyPdfiumDocumentBackend = getattr(
        pypdfium_backend_module, "PyPdfiumDocumentBackend"
    )

    pipeline_options = PdfPipelineOptions(
        do_ocr=config.do_ocr,
        do_table_structure=config.do_table_structure,
        force_backend_text=config.force_backend_text,
        ocr_options=_build_ocr_options(config),
        layout_options=_build_layout_options(config),
        table_structure_options=TableStructureOptions(
            do_cell_matching=config.table_cell_matching
        ),
    )
    backend_map = {
        "docling_parse": DoclingParseDocumentBackend,
        "pypdfium2": PyPdfiumDocumentBackend,
    }
    pdf_format_option = PdfFormatOption(
        pipeline_options=pipeline_options,
        backend=backend_map[config.pdf_backend],
    )

    # Build format map without constructing a throwaway DocumentConverter.
    # Docling's DocumentConverter accepts a partial format_options dict and
    # fills remaining formats from its defaults when omitted formats are needed;
    # we only override PDF here.
    format_options = {InputFormat.PDF: pdf_format_option}
    return DocumentConverter(format_options=format_options)