import logging import re import warnings from io import BytesIO from pathlib import Path from typing import List, Union, cast from docling_core.transforms.deserializer.doclang import DocLangDocDeserializer from docling_core.types.doc import ( ContentLayer, DocItem, DoclingDocument, ImageRef, PictureItem, ProvenanceItem, TextItem, ) from docling_core.types.doc.base import ( BoundingBox, Size, ) from docling_core.types.doc.document import DocTagsDocument from PIL import Image as PILImage from typing_extensions import override from docling.backend.abstract_backend import ( AbstractDocumentBackend, DeclarativeDocumentBackend, ) from docling.backend.docling_parse_backend import ThreadedDoclingParseDocumentBackend from docling.backend.html_backend import HTMLDocumentBackend from docling.backend.md_backend import MarkdownDocumentBackend from docling.backend.pdf_backend import PdfDocumentBackend from docling.datamodel.base_models import ( ConversionStatus, DoclingComponentType, ErrorItem, FailureCategory, InputFormat, Page, VlmStopReason, ) from docling.datamodel.document import ConversionResult, InputDocument from docling.datamodel.pipeline_options import ( VlmConvertOptions, VlmPipelineOptions, ) from docling.datamodel.pipeline_options_vlm_model import ( ApiVlmOptions, BaseVlmOptions, InferenceFramework, InlineVlmOptions, ResponseFormat, ) # VlmResponseFormat is actually ResponseFormat from pipeline_options_vlm_model # No need to import it separately as it's already imported above from docling.models.stages.vlm_convert.vlm_convert_model import VlmConvertModel from docling.models.vlm_pipeline_models.api_vlm_model import ApiVlmModel from docling.models.vlm_pipeline_models.hf_transformers_model import ( HuggingFaceTransformersVlmModel, ) from docling.models.vlm_pipeline_models.mlx_model import HuggingFaceMlxModel from docling.pipeline.base_pipeline import PaginatedPipeline from docling.utils.deepseekocr_utils import parse_deepseekocr_markdown from docling.utils.profiling import ProfilingScope, TimeRecorder _log = logging.getLogger(__name__) _DOCLANG_OPEN_RE = re.compile(r"]*)?>") def _raise_if_unsupported_threaded_backend( backend: AbstractDocumentBackend, pipeline_name: str ) -> None: if isinstance(backend, ThreadedDoclingParseDocumentBackend): raise RuntimeError( f"{pipeline_name} does not support ThreadedDoclingParseDocumentBackend yet. " "It still requires ordered/random page access via load_page() and cannot " "consume iterator-only or out-of-order page delivery. Use StandardPdfPipeline instead." ) class VlmPipeline(PaginatedPipeline): def __init__(self, pipeline_options: VlmPipelineOptions): super().__init__(pipeline_options) self.keep_backend = True self.pipeline_options: VlmPipelineOptions # Check if using new VlmConvertOptions if isinstance(pipeline_options.vlm_options, VlmConvertOptions): self._initialize_new_runtime_system(pipeline_options) else: self._initialize_legacy_vlm_models(pipeline_options) self.enrichment_pipe: list = [ # Other models working on `NodeItem` elements in the DoclingDocument ] def _initialize_new_runtime_system( self, pipeline_options: VlmPipelineOptions ) -> None: """Initialize pipeline with new VlmConvertOptions and runtime system. Args: pipeline_options: Pipeline configuration with VlmConvertOptions """ vlm_convert_options = cast(VlmConvertOptions, pipeline_options.vlm_options) # Determine response format from model spec response_format = vlm_convert_options.model_spec.response_format # force_backend_text = False - use text that is coming from VLM response # force_backend_text = True - get text from backend using bounding boxes predicted by SmolDocling doctags self.force_backend_text = ( vlm_convert_options.force_backend_text and response_format == ResponseFormat.DOCTAGS ) self.keep_images = self.pipeline_options.generate_page_images # Use new VlmConvertModel stage self.build_pipe = [ VlmConvertModel( enabled=True, enable_remote_services=self.pipeline_options.enable_remote_services, artifacts_path=self.artifacts_path, options=vlm_convert_options, accelerator_options=self.pipeline_options.accelerator_options, ), ] _log.info("Using new VlmConvertModel with runtime system") def _initialize_legacy_vlm_models( self, pipeline_options: VlmPipelineOptions ) -> None: """Initialize pipeline with legacy InlineVlmOptions or ApiVlmOptions. Args: pipeline_options: Pipeline configuration with legacy VLM options Note: This method is deprecated and will be removed in a future version. """ # Legacy path - using old InlineVlmOptions or ApiVlmOptions warnings.warn( "Using legacy VLM options (InlineVlmOptions/ApiVlmOptions) is deprecated. " "Please migrate to VlmConvertOptions with preset system. " "Example: VlmConvertOptions.from_preset('smoldocling')", DeprecationWarning, stacklevel=3, ) # force_backend_text = False - use text that is coming from VLM response # force_backend_text = True - get text from backend using bounding boxes predicted by SmolDocling doctags self.force_backend_text = ( pipeline_options.force_backend_text and pipeline_options.vlm_options.response_format == ResponseFormat.DOCTAGS # type: ignore[union-attr] ) self.keep_images = self.pipeline_options.generate_page_images if isinstance(pipeline_options.vlm_options, ApiVlmOptions): self.build_pipe = [ ApiVlmModel( enabled=True, enable_remote_services=self.pipeline_options.enable_remote_services, vlm_options=cast(ApiVlmOptions, self.pipeline_options.vlm_options), ), ] elif isinstance(self.pipeline_options.vlm_options, InlineVlmOptions): vlm_options = cast(InlineVlmOptions, self.pipeline_options.vlm_options) if vlm_options.inference_framework == InferenceFramework.MLX: self.build_pipe = [ HuggingFaceMlxModel( enabled=True, artifacts_path=self.artifacts_path, accelerator_options=pipeline_options.accelerator_options, vlm_options=vlm_options, ), ] elif vlm_options.inference_framework == InferenceFramework.TRANSFORMERS: self.build_pipe = [ HuggingFaceTransformersVlmModel( enabled=True, artifacts_path=self.artifacts_path, accelerator_options=pipeline_options.accelerator_options, vlm_options=vlm_options, ), ] elif vlm_options.inference_framework == InferenceFramework.VLLM: from docling.models.vlm_pipeline_models.vllm_model import VllmVlmModel self.build_pipe = [ VllmVlmModel( enabled=True, artifacts_path=self.artifacts_path, accelerator_options=pipeline_options.accelerator_options, vlm_options=vlm_options, ), ] else: raise ValueError( f"Could not instantiate the right type of VLM pipeline: {vlm_options.inference_framework}" ) def initialize_page(self, conv_res: ConversionResult, page: Page) -> Page: with TimeRecorder(conv_res, "page_init"): images_scale = self.pipeline_options.images_scale if images_scale is not None: page._default_image_scale = images_scale _raise_if_unsupported_threaded_backend( conv_res.input._backend, self.__class__.__name__ ) page._backend = conv_res.input._backend.load_page(page.page_no - 1) # type: ignore if page._backend is not None and page._backend.is_valid(): page.size = page._backend.get_size() if self.force_backend_text: page.parsed_page = page._backend.get_segmented_page() return page def extract_text_from_backend( self, page: Page, bbox: Union[BoundingBox, None] ) -> str: # Convert bounding box normalized to 0-100 into page coordinates for cropping text = "" if bbox: if page.size: if page._backend: text = page._backend.get_text_in_rect(bbox) return text @override def _determine_status(self, conv_res: ConversionResult) -> ConversionStatus: """Determine conversion status accounting for VLM stop reasons. Extends the base implementation to detect partial failures from VLM inference, such as truncated output (LENGTH) or filtered content (CONTENT_FILTERED). """ status = super()._determine_status(conv_res) for page in conv_res.pages: vlm_response = page.predictions.vlm_response if vlm_response is None: conv_res.errors.append( ErrorItem( component_type=DoclingComponentType.PIPELINE, module_name=self.__class__.__name__, error_message="No VLM prediction.", category=FailureCategory.INFERENCE_FAILURE, page_no=page.page_no, ) ) status = ConversionStatus.PARTIAL_SUCCESS elif vlm_response.stop_reason in ( VlmStopReason.LENGTH, VlmStopReason.CONTENT_FILTERED, ): conv_res.errors.append( ErrorItem( component_type=DoclingComponentType.PIPELINE, module_name=self.__class__.__name__, error_message="VLM output incomplete " f"(stop_reason={vlm_response.stop_reason.value}).", category=FailureCategory.INFERENCE_FAILURE, page_no=page.page_no, ) ) status = ConversionStatus.PARTIAL_SUCCESS return status def _assemble_document(self, conv_res: ConversionResult) -> ConversionResult: with TimeRecorder(conv_res, "doc_assemble", scope=ProfilingScope.DOCUMENT): # Determine response format from options if isinstance(self.pipeline_options.vlm_options, VlmConvertOptions): response_format = ( self.pipeline_options.vlm_options.model_spec.response_format ) # Response format is already ResponseFormat, no mapping needed response_format_legacy = response_format else: # Legacy path response_format_legacy = ( self.pipeline_options.vlm_options.response_format ) if response_format_legacy == ResponseFormat.DOCTAGS: conv_res.document = self._turn_dt_into_doc(conv_res) elif response_format_legacy == ResponseFormat.DOCLANG: conv_res.document = self._turn_doclang_into_doc(conv_res) elif response_format_legacy == ResponseFormat.DEEPSEEKOCR_MARKDOWN: conv_res.document = self._parse_deepseekocr_markdown(conv_res) elif response_format_legacy == ResponseFormat.MARKDOWN: conv_res.document = self._convert_text_with_backend( conv_res, InputFormat.MD, MarkdownDocumentBackend ) elif response_format_legacy == ResponseFormat.HTML: conv_res.document = self._convert_text_with_backend( conv_res, InputFormat.HTML, HTMLDocumentBackend ) elif response_format_legacy == ResponseFormat.CHANDRA_HTML: conv_res.document = self._parse_chandra_html(conv_res) elif response_format_legacy == ResponseFormat.DOTS_JSON: conv_res.document = self._parse_dots_json(conv_res) else: raise RuntimeError( f"Unsupported VLM response format {response_format_legacy}" ) # Generate images of the requested element types if self.pipeline_options.generate_picture_images: scale = self.pipeline_options.images_scale for element, _level in conv_res.document.iterate_items(): if not isinstance(element, DocItem) or len(element.prov) == 0: continue if ( isinstance(element, PictureItem) and self.pipeline_options.generate_picture_images ): page_ix = element.prov[0].page_no - 1 page = conv_res.pages[page_ix] assert page.size is not None assert page.image is not None crop_bbox = ( element.prov[0] .bbox.scaled(scale=scale) .to_top_left_origin(page_height=page.size.height * scale) ) cropped_im = page.image.crop(crop_bbox.as_tuple()) element.image = ImageRef.from_pil( cropped_im, dpi=int(72 * scale) ) return conv_res def _extract_doclang_fragment(self, text: str) -> str | None: """Extract the first ... fragment from text.""" if not text or not _DOCLANG_OPEN_RE.search(text): return None start = text.find("", start) if end < 0: return None return text[start : end + len("")] def _turn_doclang_into_doc(self, conv_res: ConversionResult) -> DoclingDocument: deserializer = DocLangDocDeserializer() doclang_strings: list[str] = [] images: list[PILImage.Image] = [] # Process ALL pages in order, like DOCTAGS does for page in conv_res.pages: doclang_text = "" img = PILImage.new("RGB", (1, 1), "rgb(255,255,255)") if page.predictions.vlm_response: fragment = self._extract_doclang_fragment( page.predictions.vlm_response.text ) if fragment: doclang_text = fragment else: conv_res.errors.append( ErrorItem( component_type=DoclingComponentType.PIPELINE, module_name=self.__class__.__name__, error_message="No XML fragment found in VLM response.", category=FailureCategory.INFERENCE_FAILURE, page_no=page.page_no, ) ) conv_res.status = ConversionStatus.PARTIAL_SUCCESS if page.image: img = page.image doclang_strings.append(doclang_text) images.append(img) # Deserialize each page and attach its image page_docs: list[DoclingDocument] = [] for idx, (doclang_text, img) in enumerate(zip(doclang_strings, images)): if not doclang_text or not doclang_text.strip(): # Create empty document for missing pages empty_doc = DoclingDocument(name=f"page_{idx}") empty_doc.add_page( page_no=idx + 1, size=Size(width=img.width, height=img.height), image=ImageRef.from_pil(image=img, dpi=72), ) page_docs.append(empty_doc) continue try: page_doc = deserializer.deserialize_str(doclang_text) # Attach the image to the deserialized page page_nos = list(page_doc.pages.keys()) if page_nos: page_no = page_nos[0] page_doc.pages[page_no].image = ImageRef.from_pil(image=img, dpi=72) page_docs.append(page_doc) except Exception as exc: conv_res.errors.append( ErrorItem( component_type=DoclingComponentType.PIPELINE, module_name=self.__class__.__name__, error_message=f"DoclangDeserializer failed: {exc}", category=FailureCategory.BACKEND_FAILURE, page_no=idx + 1, ) ) conv_res.status = ConversionStatus.PARTIAL_SUCCESS # Create empty document for failed pages empty_doc = DoclingDocument(name=f"page_{idx}") empty_doc.add_page( page_no=idx + 1, size=Size(width=img.width, height=img.height), image=ImageRef.from_pil(image=img, dpi=72), ) page_docs.append(empty_doc) if not page_docs: raise RuntimeError("No pages to process.") if len(page_docs) == 1: return page_docs[0] return DoclingDocument.concatenate(docs=page_docs) def _turn_dt_into_doc(self, conv_res) -> DoclingDocument: doctags_list = [] image_list = [] for page in conv_res.pages: predicted_doctags = "" img = PILImage.new("RGB", (1, 1), "rgb(255,255,255)") if page.predictions.vlm_response: predicted_doctags = page.predictions.vlm_response.text if page.image: img = page.image image_list.append(img) doctags_list.append(predicted_doctags) doctags_list_c = cast(List[Union[Path, str]], doctags_list) image_list_c = cast(List[Union[Path, PILImage.Image]], image_list) doctags_doc = DocTagsDocument.from_doctags_and_image_pairs( doctags_list_c, image_list_c ) conv_res.document = DoclingDocument.load_from_doctags( doctag_document=doctags_doc ) # If forced backend text, replace model predicted text with backend one if page.size: if self.force_backend_text: scale = self.pipeline_options.images_scale for element, _level in conv_res.document.iterate_items(): if not isinstance(element, TextItem) or len(element.prov) == 0: continue crop_bbox = ( element.prov[0] .bbox.scaled(scale=scale) .to_top_left_origin(page_height=page.size.height * scale) ) txt = self.extract_text_from_backend(page, crop_bbox) element.text = txt element.orig = txt return conv_res.document def _parse_deepseekocr_markdown( self, conv_res: ConversionResult ) -> DoclingDocument: """Parse DeepSeek OCR markdown with label[[x1, y1, x2, y2]] format. Labels supported: - text: Standard body text - title: Main document or section titles - sub_title: Secondary headings or sub-headers - table: Tabular data - table_caption: Descriptive text for tables - figure: Image-based elements or diagrams - figure_caption: Titles or descriptions for figures/images - header / footer: Content at top or bottom margins of pages """ page_docs = [] for pg_idx, page in enumerate(conv_res.pages): predicted_text = "" if page.predictions.vlm_response: predicted_text = page.predictions.vlm_response.text assert page.size is not None # Parse single page using the utility function # Pass vlm_options.scale to convert bboxes from scaled image coords to original PDF coords page_doc = parse_deepseekocr_markdown( content=predicted_text, original_page_size=page.size, page_no=pg_idx + 1, filename=conv_res.input.file.name or "file", page_image=page.image, ) page_docs.append(page_doc) # Add page metadata and concatenate return self._add_page_metadata_and_concatenate(page_docs, conv_res) def _parse_chandra_html(self, conv_res: ConversionResult) -> DoclingDocument: """Parse chandra-ocr-2 HTML output into a DoclingDocument.""" from docling.utils.chandra_utils import parse_chandra_html page_docs = [] for pg_idx, page in enumerate(conv_res.pages): predicted_text = "" if page.predictions.vlm_response: predicted_text = page.predictions.vlm_response.text assert page.size is not None page_doc = parse_chandra_html( content=predicted_text, original_page_size=page.size, page_no=pg_idx + 1, filename=conv_res.input.file.name or "file", page_image=page.image, ) page_docs.append(page_doc) return self._add_page_metadata_and_concatenate(page_docs, conv_res) def _parse_dots_json(self, conv_res: ConversionResult) -> DoclingDocument: """Parse dots.ocr / dots.mocr JSON output into a DoclingDocument.""" from docling.utils.dots_utils import parse_dots_json from docling.utils.vlm_utils import compute_qwen2vl_image_size vlm_options = self.pipeline_options.vlm_options if isinstance(vlm_options, (VlmConvertOptions, BaseVlmOptions)): vlm_scale = vlm_options.scale vlm_max_size = vlm_options.max_size else: raise TypeError( "DOTS JSON parsing requires VlmConvertOptions or BaseVlmOptions, " f"got {type(vlm_options).__name__}." ) page_docs = [] for pg_idx, page in enumerate(conv_res.pages): predicted_text = "" if page.predictions.vlm_response: predicted_text = page.predictions.vlm_response.text assert page.size is not None inference_image = page.get_image(scale=vlm_scale, max_size=vlm_max_size) model_image_size = None if inference_image is not None: model_image_size = compute_qwen2vl_image_size( width=inference_image.width, height=inference_image.height, scale=1.0, max_size=None, ) page_doc = parse_dots_json( content=predicted_text, original_page_size=page.size, page_no=pg_idx + 1, filename=conv_res.input.file.name or "file", page_image=page.image, model_image_size=model_image_size, ) page_docs.append(page_doc) return self._add_page_metadata_and_concatenate(page_docs, conv_res) def _extract_code_block(self, text: str) -> str: """ Extracts text from markdown code blocks (enclosed in triple backticks). If no code blocks are found, returns the original text. Args: text (str): Input text that may contain markdown code blocks Returns: str: Extracted code if code blocks exist, otherwise original text """ # Regex pattern to match content between triple backticks # This handles multiline content and optional language specifier pattern = r"^```(?:\w*\n)?(.*?)```(\n)*$" # Search with DOTALL flag to match across multiple lines mtch = re.search(pattern, text, re.DOTALL) if mtch: # Return only the content of the first capturing group return mtch.group(1) else: # No code blocks found, return original text return text def _add_page_metadata_and_concatenate( self, page_docs: List[DoclingDocument], conv_res: ConversionResult, ) -> DoclingDocument: """ Add page metadata to page documents and concatenate them. Args: page_docs: List of page documents to process conv_res: Conversion result containing page information Returns: DoclingDocument: Concatenated document with page metadata """ for pg_idx, (page_doc, page) in enumerate(zip(page_docs, conv_res.pages)): # Add page metadata to the page document before concatenation if page.image is not None: pg_width = page.image.width pg_height = page.image.height else: pg_width = 1 pg_height = 1 page_no = pg_idx + 1 if page_no not in page_doc.pages: page_doc.add_page( page_no=page_no, size=Size(width=pg_width, height=pg_height), image=ImageRef.from_pil(image=page.image, dpi=72) if page.image else None, ) # Concatenate all page documents to preserve hierarchy return DoclingDocument.concatenate(docs=page_docs) def _convert_text_with_backend( self, conv_res: ConversionResult, input_format: InputFormat, backend_class: type[DeclarativeDocumentBackend], ) -> DoclingDocument: """ Convert text-based formats (Markdown, HTML) into DoclingDocument using a backend. Args: conv_res: The conversion result containing pages with VLM predictions input_format: The format type (MD or HTML) backend_class: The backend class to use for conversion Returns: DoclingDocument: The assembled document """ page_docs = [] for pg_idx, page in enumerate(conv_res.pages): predicted_text = "" if page.predictions.vlm_response: predicted_text = page.predictions.vlm_response.text + "\n\n" # Extract content from code blocks if present predicted_text = self._extract_code_block(text=predicted_text) # Convert text to document using specified backend response_bytes = BytesIO(predicted_text.encode("utf8")) out_doc = InputDocument( path_or_stream=response_bytes, filename=conv_res.input.file.name, format=input_format, backend=backend_class, ) backend = backend_class( in_doc=out_doc, path_or_stream=response_bytes, ) page_doc = backend.convert() # Modify provenance in place for all items in the page document for item, level in page_doc.iterate_items( with_groups=True, traverse_pictures=True, included_content_layers=set(ContentLayer), ): if isinstance(item, DocItem): item.prov = [ ProvenanceItem( page_no=pg_idx + 1, bbox=BoundingBox( t=0.0, b=0.0, l=0.0, r=0.0 ), # FIXME: would be nice not to have to "fake" it charspan=[0, 0], ) ] page_docs.append(page_doc) # Add page metadata and concatenate return self._add_page_metadata_and_concatenate(page_docs, conv_res) @classmethod def get_default_options(cls) -> VlmPipelineOptions: return VlmPipelineOptions() @classmethod def is_backend_supported(cls, backend: AbstractDocumentBackend): return isinstance(backend, PdfDocumentBackend)