feat(ocr): Add parallel batch OCR processing via ThreadPoolExecutor

packages/markitdown-ocr/src/markitdown_ocr/_docx_converter_with_ocr.py

@@ -127,31 +127,39 @@ def _extract_and_ocr_images(
         self, file_stream: BinaryIO, ocr_service: LLMVisionOCRService
     ) -> dict[str, str]:
         """
-        Extract images from DOCX and OCR them.
+        Extract images from DOCX and OCR them in parallel.
 
         Returns:
             Dict mapping image relationship IDs to raw OCR text (no markers).
         """
-        ocr_map = {}
+        ocr_map: dict[str, str] = {}
 
         try:
             file_stream.seek(0)
             doc = Document(file_stream)
 
+            # Phase 1: Collect all image streams with their rIds
+            image_specs: list[tuple[str, BinaryIO]] = []  # (rId, stream)
             for rel in doc.part.rels.values():
                 if "image" in rel.target_ref.lower():
                     try:
                         image_bytes = rel.target_part.blob
                         image_stream = io.BytesIO(image_bytes)
-                        ocr_result = ocr_service.extract_text(image_stream)
-
-                        if ocr_result.text.strip():
-                            # Store raw text only — markers added later
-                            ocr_map[rel.rId] = ocr_result.text.strip()
-
+                        image_specs.append((rel.rId, image_stream))
                     except Exception:
                         continue
 
+            # Phase 2: Batch OCR all images in parallel
+            if image_specs:
+                rids = [spec[0] for spec in image_specs]
+                streams = [spec[1] for spec in image_specs]
+                results = ocr_service.extract_text_batch(
+                    [(s, None) for s in streams]
+                )
+                for rId, result in zip(rids, results):
+                    if result.text.strip():
+                        ocr_map[rId] = result.text.strip()
+
         except Exception:
             pass
 

packages/markitdown-ocr/src/markitdown_ocr/_ocr_service.py

@@ -4,6 +4,7 @@
 """
 
 import base64
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from typing import Any, BinaryIO
 from dataclasses import dataclass
 
@@ -28,6 +29,7 @@ def __init__(
         client: Any,
         model: str,
         default_prompt: str | None = None,
+        max_workers: int = 5,
     ) -> None:
         """
         Initialize LLM Vision OCR service.
@@ -36,9 +38,11 @@ def __init__(
             client: OpenAI-compatible client
             model: Model name (e.g., 'gpt-4o', 'gemini-2.0-flash')
             default_prompt: Default prompt for OCR extraction
+            max_workers: Maximum number of parallel OCR workers (default 5)
         """
         self.client = client
         self.model = model
+        self.max_workers = max_workers
         self.default_prompt = default_prompt or (
             "Extract all text from this image. "
             "Return ONLY the extracted text, maintaining the original "
@@ -108,3 +112,59 @@ def extract_text(
             return OCRResult(text="", backend_used="llm_vision", error=str(e))
         finally:
             image_stream.seek(0)
+
+    def extract_text_batch(
+        self,
+        images: list[tuple[BinaryIO, StreamInfo | None]],
+        prompt: str | None = None,
+        **kwargs: Any,
+    ) -> list[OCRResult]:
+        """
+        Extract text from multiple images in parallel using a thread pool.
+
+        Args:
+            images: List of (image_stream, stream_info) tuples
+            prompt: Optional prompt override (shared across all images)
+            **kwargs: Additional arguments (max_workers override, etc.)
+
+        Returns:
+            List of OCRResult, one per input image (same order as input).
+            Individual failures are captured in the OCRResult's error field
+            rather than raising an exception.
+        """
+        if not images:
+            return []
+
+        max_workers = kwargs.get("max_workers", self.max_workers)
+
+        def _process_one(
+            idx: int, img_stream: BinaryIO, stream_info: StreamInfo | None
+        ) -> tuple[int, OCRResult]:
+            """Wrapper that catches all exceptions so one failure doesn't
+            kill the entire batch."""
+            try:
+                result = self.extract_text(
+                    img_stream, prompt=prompt, stream_info=stream_info
+                )
+            except Exception as e:
+                result = OCRResult(
+                    text="", backend_used="llm_vision", error=str(e)
+                )
+            return idx, result
+
+        results: list[OCRResult | None] = [None] * len(images)
+
+        with ThreadPoolExecutor(max_workers=max_workers) as executor:
+            futures = {
+                executor.submit(_process_one, i, img, info): i
+                for i, (img, info) in enumerate(images)
+            }
+            for future in as_completed(futures):
+                idx, result = future.result()
+                results[idx] = result
+
+        # Defensive: any slot still None → empty result
+        return [
+            r if r is not None else OCRResult(text="", backend_used="llm_vision")
+            for r in results
+        ]