feat: add dependency caching for local OCR engine, enhance observability logging, and improve startup configuration

Author: fbkaragoz

ocr_pipeline/Dockerfile

@@ -20,4 +20,4 @@ ENV PYTHONPATH=/app
 
 EXPOSE 39672
 
-CMD ["uvicorn", "ocr_pipeline.main:app", "--host", "0.0.0.0", "--port", "39672"]
+CMD ["uvicorn", "ocr_pipeline.main:app", "--host", "0.0.0.0", "--port", "39672", "--no-access-log"]

ocr_pipeline/Dockerfile.cpu

@@ -31,4 +31,4 @@ ENV LOCAL_DEVICE=cpu
 
 EXPOSE 39672
 
-CMD ["uvicorn", "ocr_pipeline.main:app", "--host", "0.0.0.0", "--port", "39672"]
+CMD ["uvicorn", "ocr_pipeline.main:app", "--host", "0.0.0.0", "--port", "39672", "--no-access-log"]

ocr_pipeline/config.py

@@ -59,6 +59,7 @@ class Settings(BaseSettings):
     # Server
     host: str = "0.0.0.0"
     port: int = 39672
+    log_level: str = "INFO"
 
     # Pipeline
     pipeline_version: str = "2.0.0"

ocr_pipeline/main.py

@@ -25,9 +25,11 @@
 from ocr_pipeline.services.startup import wait_for_model_server
 
 logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s [%(name)s] %(levelname)s: %(message)s",
+    level=getattr(logging, settings.log_level.upper(), logging.INFO),
+    format="%(asctime)s %(levelname)-7s %(name)s :: %(message)s",
 )
+for noisy_logger in ("httpx", "httpcore", "urllib3", "huggingface_hub"):
+    logging.getLogger(noisy_logger).setLevel(logging.WARNING)
 logger = logging.getLogger("ocr_pipeline")
 
 

ocr_pipeline/services/local_ocr_engine.py

@@ -97,29 +97,41 @@ def __init__(self, model_name: str | None = None) -> None:
         self._device: str | None = None
         self._dtype: Any = None
         self._lock = asyncio.Lock()
+        self._load_error: BaseException | None = None
         self._initialized = True
 
     async def _ensure_loaded(self) -> None:
         if self._model is not None:
             return
+        if self._load_error is not None:
+            raise RuntimeError("Local OCR engine failed to load") from self._load_error
         async with self._lock:
             if self._model is not None:
                 return
-            await asyncio.to_thread(self._load_blocking)
+            if self._load_error is not None:
+                raise RuntimeError(
+                    "Local OCR engine failed to load"
+                ) from self._load_error
+            try:
+                await asyncio.to_thread(self._load_blocking)
+            except Exception as exc:
+                self._load_error = exc
+                logger.error("Local OCR engine failed to load: %s", exc)
+                raise
 
     def _load_blocking(self) -> None:
-        if find_spec("torch") is None:
+        missing = [
+            package
+            for package in ("torch", "transformers", "easydict")
+            if find_spec(package) is None
+        ]
+        if missing:
             raise RuntimeError(
-                "MODEL_BACKEND=local requires `transformers` and `torch`. "
-                "Install with: pip install -r requirements-local.txt"
+                "MODEL_BACKEND=local missing package(s): "
+                f"{', '.join(missing)}. Install with: "
+                "pip install -r ocr_pipeline/requirements.txt -r requirements-local.txt"
             )
-        try:
-            from transformers import AutoModel, AutoTokenizer
-        except ImportError as exc:
-            raise RuntimeError(
-                "MODEL_BACKEND=local requires `transformers` and `torch`. "
-                "Install with: pip install -r requirements-local.txt"
-            ) from exc
+        from transformers import AutoModel, AutoTokenizer
 
         device = _resolve_device(settings.local_device)
         dtype = _resolve_dtype(settings.local_dtype, device)

ocr_pipeline/services/run_orchestrator.py

@@ -133,6 +133,13 @@ async def create_run(
         for s in staged:
             await self.db.link_run_document(run_id, s.document_id, status="pending")
 
+        logger.info(
+            "run=%s queued docs=%d pages=%d name=%s",
+            run_id,
+            len(staged),
+            pages_total,
+            name or "-",
+        )
         observability.job_created()
         return CreateRunResult(
             run_id=run_id, documents=staged, pages_total_estimate=pages_total
@@ -167,6 +174,7 @@ async def retry_incomplete_run(self, run_id: str) -> CreateRunResult:
             raise ValueError("No incomplete documents to retry")
 
         name = run.get("name") or run_id
+        logger.info("run=%s retry queued incomplete_docs=%d", run_id, len(retry_paths))
         result = await self.create_run(
             retry_paths,
             name=f"{name} retry",
@@ -189,15 +197,20 @@ async def _run(
     ) -> None:
         run_id = result.run_id
         started_at = _now()
+        pages_total = result.pages_total_estimate
+        pages_completed = 0
+        documents_meta: list = []
         await self.db.update_run(
             run_id, status="processing", stage="ocr", started_at=started_at
         )
+        logger.info(
+            "run=%s started docs=%d pages=%d",
+            run_id,
+            len(result.documents),
+            pages_total,
+        )
         await self._emit(run_id, "run_started", {"started_at": started_at})
 
-        pages_total = result.pages_total_estimate
-        pages_completed = 0
-        documents_meta: list = []
-
         async def page_event(event: dict) -> None:
             nonlocal pages_completed
             etype = event.get("type")
@@ -214,15 +227,39 @@ async def page_event(event: dict) -> None:
                     token_count=event.get("token_count", 0),
                     validation_status=event.get("validation_status", "pass"),
                 )
+                logger.info(
+                    "run=%s page=%d/%d doc=%s status=%s time=%.1fms",
+                    run_id,
+                    pages_completed,
+                    pages_total,
+                    event.get("document"),
+                    event.get("validation_status"),
+                    event.get("processing_time_ms", 0.0),
+                )
             elif etype == "page_retry":
                 observability.page_retry()
+                logger.info(
+                    "run=%s page=%s retry attempt=%s strategy=%s reason=%s",
+                    run_id,
+                    event.get("page"),
+                    event.get("attempt"),
+                    event.get("new_strategy"),
+                    event.get("reason"),
+                )
             await self._emit(run_id, etype, event)
 
         try:
-            for staged in result.documents:
+            for index, staged in enumerate(result.documents, start=1):
                 paths = self.storage.artifact_paths(
                     run_id, staged.document_id, staged.filename
                 )
+                logger.info(
+                    "run=%s doc=%d/%d started %s",
+                    run_id,
+                    index,
+                    len(result.documents),
+                    staged.filename,
+                )
                 processor = BatchProcessor(
                     self.db, event_callback=page_event, strip_refs=strip_refs
                 )
@@ -238,6 +275,16 @@ async def page_event(event: dict) -> None:
                 await self.db.update_run(
                     run_id, documents_completed=len(documents_meta)
                 )
+                logger.info(
+                    "run=%s doc=%d/%d completed %s pass=%d warn=%d fail=%d",
+                    run_id,
+                    index,
+                    len(result.documents),
+                    staged.filename,
+                    doc_meta.pages_pass,
+                    doc_meta.pages_warn,
+                    doc_meta.pages_fail,
+                )
 
             dataset_bundle = await self._maybe_export(
                 run_id, documents_meta, export_parquet
@@ -253,6 +300,7 @@ async def page_event(event: dict) -> None:
                 dataset_bundle=dataset_bundle,
                 completed_at=completed_at,
             )
+            logger.info("run=%s completed bundle=%s", run_id, dataset_bundle or "-")
             observability.job_completed()
             await self._emit(
                 run_id,
@@ -285,6 +333,7 @@ async def _maybe_export(
         if not (export_parquet and documents_meta):
             return None
         await self.db.update_run(run_id, stage="exporting")
+        logger.info("run=%s exporting dataset docs=%d", run_id, len(documents_meta))
         await self._emit(run_id, "dataset_export_started", {})
         exports = []
         for did, paths, meta in documents_meta:

requirements-local.txt

@@ -10,3 +10,4 @@ transformers>=4.46.0
 accelerate>=0.34.0
 einops>=0.8.0
 sentencepiece>=0.2.0
+easydict>=1.13

scripts/start.sh

@@ -31,9 +31,20 @@ mkdir -p "$INPUT_DIR" "$OUTPUT_DIR"
 export PYTHONPATH="${PYTHONPATH:-}:$(pwd)"
 PORT="${PORT:-39672}"
 HOST="${HOST:-0.0.0.0}"
+if [[ -z "${PYTHON_BIN:-}" && -x ".venv/bin/python" ]]; then
+  PYTHON_BIN=".venv/bin/python"
+else
+  PYTHON_BIN="${PYTHON_BIN:-python3}"
+fi
 
 echo "→ OpenCR  backend=$MODEL_BACKEND  http://$HOST:$PORT"
 echo "  input=$INPUT_DIR"
 echo "  output=$OUTPUT_DIR"
+echo "  python=$PYTHON_BIN"
+
+UVICORN_ARGS=()
+if [[ "${ACCESS_LOG:-0}" != "1" ]]; then
+  UVICORN_ARGS+=(--no-access-log)
+fi
 
-exec python3 -m uvicorn ocr_pipeline.main:app --host "$HOST" --port "$PORT" "$@"
+exec "$PYTHON_BIN" -m uvicorn ocr_pipeline.main:app --host "$HOST" --port "$PORT" "${UVICORN_ARGS[@]}" "$@"

tests/test_local_ocr_engine.py

@@ -0,0 +1,29 @@
+import asyncio
+
+from ocr_pipeline.services.local_ocr_engine import LocalOCREngine
+
+
+def test_local_engine_caches_load_failure(monkeypatch):
+    async def _scenario():
+        LocalOCREngine._instance = None
+        engine = LocalOCREngine()
+        calls = 0
+
+        def fail_load():
+            nonlocal calls
+            calls += 1
+            raise RuntimeError("missing dependency")
+
+        monkeypatch.setattr(engine, "_load_blocking", fail_load)
+
+        for _ in range(2):
+            try:
+                await engine._ensure_loaded()
+            except RuntimeError:
+                pass
+            else:
+                raise AssertionError("expected load failure")
+
+        assert calls == 1
+
+    asyncio.run(_scenario())

tests/test_requirements.py

@@ -0,0 +1,8 @@
+from pathlib import Path
+
+
+def test_local_requirements_include_deepseek_remote_code_dependencies():
+    requirements = (Path(__file__).parents[1] / "requirements-local.txt").read_text(
+        encoding="utf-8"
+    )
+    assert "easydict" in requirements