Implement Datalab-compatible API with async endpoints and deepseek OCR backend

- New endpoints: /api/v1/convert, /api/v1/extract, /api/v1/segment,
  /api/v1/ocr, /api/v1/table_rec, /api/v1/marker (deprecated),
  /api/v1/create-document, /api/v1/files/*, /api/v1/thumbnails
- Async submit-and-poll pattern matching Datalab API spec
- X-API-Key authentication via API_KEY env var
- Filesystem-based request store for multi-worker support
- OCR_BACKEND=deepseek mode: uses deepseek-ocr via ollama for OCR
  (prompt: <|grounding|>Free OCR.)
- build_options() now propagates output_format to parsed_opts
- Fixed docs page CSS curly brace conflict with .format()
- Renamed marker_endpoint_referece.md -> marker_endpoint_reference.md
- Containerfile: added poppler-utils, pandoc; 3 gunicorn workers
- gunicorn.conf: post_fork hook for background worker threads

Containerfile

@@ -10,7 +10,8 @@ FROM rocm/pytorch:rocm7.2.4_ubuntu24.04_py3.12_pytorch_release_2.9.1
 RUN apt-get update && \
     apt-get install -y --no-install-recommends \
         curl ca-certificates tini procps git gcc g++ zlib1g-dev libjpeg-dev \
-        libpango-1.0-0 libharfbuzz0b libpangoft2-1.0-0 && \
+        libpango-1.0-0 libharfbuzz0b libpangoft2-1.0-0 \
+        poppler-utils pandoc && \
     apt-get remove -y python3-blinker || true && \
     rm -rf /var/lib/apt/lists/*
 
@@ -39,33 +40,41 @@ RUN mkdir -p /app/marker/static/fonts && \
 
 # ---- final image ----
 COPY app.py           /app/
+COPY request_store.py /app/
+COPY deepseek_ocr.py  /app/
 COPY gunicorn.conf.py /app/
 COPY entrypoint.sh    /app/entrypoint.sh
 
 RUN chmod +x /app/entrypoint.sh
 
-ENV \
-    HOME=/app \
-    PYTHONUNBUFFERED=1 \
-    PYTHONDONTWRITEBYTECODE=1 \
-    MARKER_OUTPUT_DIR=/app/conversion_results \
-    UPLOAD_DIR=/app/uploads \
-    OLLAMA_HOST=http://localhost:11435 \
-    DEESEEK_OCR_MODEL=deepseek-ocr \
-    AMD_COMPUTE=true \
-    TORCH_DEVICE= \
-    MODEL_DTYPE=float32 \
-    PORT=8000 \
-    HOST=0.0.0.0 \
-    LLM_SERVICE=marker.services.openai.OpenAIService \
-    USE_LLM=false \
-    OPENAI_BASE_URL=http://localhost:11435/v1 \
-    OPENAI_MODEL= \
-    GUNICORN_WORKERS=1
+ENV HOME=/app
+ENV PYTHONUNBUFFERED=1
+ENV PYTHONDONTWRITEBYTECODE=1
+ENV MARKER_OUTPUT_DIR=/app/conversion_results
+ENV UPLOAD_DIR=/app/uploads
+ENV RESULTS_DIR=/app/conversion_results
+ENV SELF_URL=http://localhost:8001
+ENV OLLAMA_HOST=http://localhost:11435
+ENV DEESEEK_OCR_MODEL=deepseek-ocr
+ENV AMD_COMPUTE=true
+ENV TORCH_DEVICE=
+ENV MODEL_DTYPE=float32
+ENV PORT=8001
+ENV HOST=0.0.0.0
+ENV LLM_SERVICE=marker.services.openai.OpenAIService
+ENV USE_LLM=false
+ENV OPENAI_BASE_URL=http://localhost:11435/v1
+ENV OPENAI_MODEL=
+ENV OCR_BACKEND=marker
+ENV DEEPSEEK_OLLAMA_HOST=http://localhost:11434
+ENV DEEPSEEK_OCR_MODEL=deepseek-ocr:latest
+ENV DEEPSEEK_OCR_PROMPT="<|grounding|>Free OCR."
+ENV API_KEY=
+ENV GUNICORN_WORKERS=3
 
 USER marker
 
-EXPOSE 8000
+EXPOSE 8001
 
 ENTRYPOINT ["/app/entrypoint.sh"]
 CMD ["gunicorn", "-c", "gunicorn.conf.py", "app:app_instance"]

deepseek_ocr.py

@@ -0,0 +1,106 @@
+from __future__ import annotations
+
+import base64
+import io
+import os
+import tempfile
+import subprocess
+import requests
+from PIL import Image
+
+OLLAMA_HOST = os.environ.get("DEEPSEEK_OLLAMA_HOST", "http://localhost:11434")
+DEEPSEEK_MODEL = os.environ.get("DEEPSEEK_OCR_MODEL", "deepseek-ocr:latest")
+DEEPSEEK_PROMPT = os.environ.get("DEEPSEEK_OCR_PROMPT", "<|grounding|>Free OCR.")
+
+
+def _page_to_image(raw_bytes: bytes, page_num: int = 0) -> bytes | None:
+    with tempfile.TemporaryDirectory() as tmpdir:
+        result = subprocess.run(
+            [
+                "pdftoppm", "-png",
+                "-f", str(page_num + 1),
+                "-l", str(page_num + 1),
+                "-",
+                os.path.join(tmpdir, "page"),
+            ],
+            input=raw_bytes,
+            capture_output=True,
+            timeout=120,
+        )
+        if result.returncode != 0:
+            return None
+        out_files = os.listdir(tmpdir)
+        if not out_files:
+            return None
+        img_path = os.path.join(tmpdir, out_files[0])
+        img = Image.open(img_path)
+        buf = io.BytesIO()
+        img.save(buf, format="PNG")
+        return buf.getvalue()
+
+
+def _image_to_text(image_bytes: bytes) -> str:
+    b64 = base64.b64encode(image_bytes).decode()
+    resp = requests.post(
+        f"{OLLAMA_HOST}/api/generate",
+        json={
+            "model": DEEPSEEK_MODEL,
+            "prompt": DEEPSEEK_PROMPT,
+            "images": [b64],
+            "stream": False,
+            "options": {"temperature": 0},
+        },
+        timeout=120,
+    )
+    resp.raise_for_status()
+    return resp.json().get("response", "")
+
+
+def ocr_pdf(raw_bytes: bytes, max_pages: int | None = None, page_range: str | None = None) -> dict:
+    result = {
+        "success": True,
+        "text": "",
+        "pages": [],
+        "page_count": 0,
+    }
+    pages_to_process = _parse_page_range(page_range, max_pages, raw_bytes)
+    for page_num in pages_to_process:
+        img_bytes = _page_to_image(raw_bytes, page_num)
+        if img_bytes is None:
+            continue
+        text = _image_to_text(img_bytes)
+        result["pages"].append({"page": page_num, "text": text})
+        result["text"] += f"\n\n--- PAGE {page_num + 1} ---\n\n{text}"
+    result["page_count"] = len(result["pages"])
+    return result
+
+
+def _parse_page_range(page_range: str | None, max_pages: int | None, raw_bytes: bytes) -> list[int]:
+    import tempfile, os, subprocess
+    total = 0
+    with tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") as f:
+        f.write(raw_bytes)
+        f.flush()
+        r = subprocess.run(
+            ["pdfinfo", f.name],
+            capture_output=True, text=True, timeout=30,
+        )
+        os.unlink(f.name)
+    for line in r.stdout.splitlines():
+        if line.startswith("Pages:"):
+            total = int(line.split(":")[1].strip())
+            break
+    if page_range:
+        pages = set()
+        for part in page_range.split(","):
+            part = part.strip()
+            if "-" in part:
+                a, b = part.split("-", 1)
+                for p in range(int(a.strip()), int(b.strip()) + 1):
+                    pages.add(p)
+            else:
+                pages.add(int(part))
+        return sorted(pages)
+    if max_pages is not None:
+        return list(range(min(max_pages, total)))
+    return list(range(total))

docker-compose.yml

@@ -14,6 +14,12 @@ services:
       - MODEL_DTYPE=${MODEL_DTYPE:-float32}
       - OLLAMA_HOST=${OLLAMA_HOST:-http://localhost:11435}
       - DEESEEK_OCR_MODEL=${DEESEEK_OCR_MODEL:-}
+      - OCR_BACKEND=${OCR_BACKEND:-marker}
+      - DEEPSEEK_OLLAMA_HOST=${DEEPSEEK_OLLAMA_HOST:-http://host.containers.internal:11434}
+      - DEEPSEEK_OCR_MODEL=${DEEPSEEK_OCR_MODEL:-deepseek-ocr:latest}
+      - DEEPSEEK_OCR_PROMPT=<|grounding|>Free OCR.
+      - API_KEY=${API_KEY:-}
+      - SELF_URL=http://localhost:8001
       - LLM_SERVICE=marker.services.openai.OpenAIService
       - USE_LLM=${USE_LLM:-false}
       - OPENAI_BASE_URL=http://localhost:11435/v1
@@ -21,7 +27,7 @@ services:
       - OPENAI_MODEL=
       - PORT=${PORT:-8001}
       - HOST=0.0.0.0
-      - GUNICORN_WORKERS=${GUNICORN_WORKERS:-1}
+      - GUNICORN_WORKERS=${GUNICORN_WORKERS:-3}
       - GUNICORN_THREADS=${GUNICORN_THREADS:-2}
       - GUNICORN_TIMEOUT=${GUNICORN_TIMEOUT:-600}
     ports:

entrypoint.sh

@@ -16,5 +16,5 @@ fi
 # Check torch/ROCm availability
 python3 -c "import torch; dev=torch.device('cuda' if torch.cuda.is_available() else 'cpu'); print(f'[entrypoint] PyTorch device: {dev}, ROCm: {torch.version.rocm if hasattr(torch.version, \"rocm\") else \"n/a\"}')" 2>&1
 
-echo "[entrypoint] Starting marker-api on ${HOST:-0.0.0.0}:${PORT:-8000} ..."
+echo "[entrypoint] Starting marker-api on ${HOST:-0.0.0.0}:${PORT:-8001} ..."
 exec "$@"

gunicorn.conf.py

@@ -1,29 +1,91 @@
-# gunicorn.conf.py -- production WSGI config for marker-api
 import os
 import multiprocessing
+import threading
 
-port = int(os.environ.get("PORT", "8000"))
-workers = min(multiprocessing.cpu_count() * 2 + 1, 8)
+port = int(os.environ.get("PORT", "8001"))
+workers = int(os.environ.get("GUNICORN_WORKERS", "3"))
+threads = int(os.environ.get("GUNICORN_THREADS", "4"))
+timeout = int(os.environ.get("GUNICORN_TIMEOUT", "600"))
 
 bind = f"0.0.0.0:{port}"
-workers = int(os.environ.get("GUNICORN_WORKERS", "1"))
 workers = min(workers, 8)
 
 worker_class = "gthread"
-threads = int(os.environ.get("GUNICORN_THREADS", "4"))
-timeout = int(os.environ.get("GUNICORN_TIMEOUT", "300"))
 graceful_timeout = 60
 
 accesslog = "-"
 errorlog = "-"
 loglevel = os.environ.get("GUNICORN_LOGLEVEL", "info")
 
-# Preload so models are loaded once in master process (not duplicated per-worker)
-preload_app = True
+# Do NOT preload — each worker starts its own background thread and loads models
+preload_app = False
 
-# Enable worker recycling to prevent memory leaks
 max_requests = 1000
 max_requests_jitter = 100
 
-# Tempdir for marker uploads
 tempdir = "/app/uploads"
+
+
+def post_fork(server, worker):
+    """Start background worker thread after fork (filesystem-based store works across workers)."""
+    t = threading.Thread(target=_bg_worker, daemon=True)
+    t.start()
+
+
+def _bg_worker():
+    import app as marker_api
+    from request_store import _job_queue, _job_cond, get_request, complete_request, fail_request, get_result
+
+    while True:
+        import time
+        import traceback
+        with _job_cond:
+            while not _job_queue:
+                _job_cond.wait()
+            job = _job_queue.pop(0)
+        request_id, endpoint, raw_bytes, filename, opts = job
+        try:
+            if endpoint == "convert":
+                result = marker_api.convert_with_ocr_backend(raw_bytes, filename, **opts)
+            elif endpoint == "ocr":
+                from deepseek_ocr import ocr_pdf
+                result = ocr_pdf(raw_bytes, max_pages=opts.get("max_pages"), page_range=opts.get("page_range"))
+            elif endpoint == "extract":
+                conv_result = marker_api.convert_with_ocr_backend(raw_bytes, filename, **opts)
+                if conv_result["success"]:
+                    import json
+                    schema_json = opts.get("page_schema") or "{}"
+                    try:
+                        schema = json.loads(schema_json) if isinstance(schema_json, str) else schema_json
+                    except json.JSONDecodeError:
+                        schema = {}
+                    extraction = marker_api._apply_extraction_schema(conv_result["output"], schema)
+                    result = {**conv_result, "extraction": extraction}
+                else:
+                    result = conv_result
+            elif endpoint == "segment":
+                conv_result = marker_api.convert_with_ocr_backend(raw_bytes, filename, **opts)
+                if conv_result["success"]:
+                    import json
+                    schema_json = opts.get("segmentation_schema") or "{}"
+                    try:
+                        schema = json.loads(schema_json) if isinstance(schema_json, str) else schema_json
+                    except json.JSONDecodeError:
+                        schema = {}
+                    segments = marker_api._apply_segmentation_schema(conv_result["output"], schema)
+                    result = {**conv_result, "segments": segments}
+                else:
+                    result = conv_result
+            elif endpoint == "table_rec":
+                conv_result = marker_api.convert_with_ocr_backend(raw_bytes, filename, **opts, output_format="json")
+                tables = marker_api._extract_tables(conv_result)
+                result = {**conv_result, "tables": tables}
+            else:
+                result = {"success": False, "error": f"Unknown endpoint: {endpoint}"}
+            if result["success"]:
+                complete_request(request_id, result)
+            else:
+                fail_request(request_id, result.get("error", "Unknown error"))
+        except Exception as exc:
+            traceback.print_exc()
+            fail_request(request_id, str(exc))

request_store.py

@@ -0,0 +1,121 @@
+from __future__ import annotations
+
+import json
+import os
+import time
+import threading
+from pathlib import Path
+
+DEFAULT_RESULTS_DIR = os.environ.get("RESULTS_DIR", "/app/conversion_results")
+DEFAULT_TTL = 3600  # 1 hour (matches Datalab)
+
+_lock = threading.Lock()
+
+
+def _req_path(request_id: str) -> Path:
+    return Path(DEFAULT_RESULTS_DIR) / request_id
+
+
+def create_request(request_id: str, metadata: dict | None = None) -> dict:
+    d = _req_path(request_id)
+    d.mkdir(parents=True, exist_ok=True)
+    entry = {
+        "request_id": request_id,
+        "status": "processing",
+        "success": None,
+        "created_at": time.time(),
+        "metadata": metadata or {},
+    }
+    with _lock:
+        (d / "status.json").write_text(json.dumps(entry))
+    return entry
+
+
+def get_request(request_id: str) -> dict | None:
+    p = _req_path(request_id) / "status.json"
+    if not p.exists():
+        return None
+    with _lock:
+        return json.loads(p.read_text())
+
+
+def update_request(request_id: str, **kwargs) -> dict | None:
+    d = _req_path(request_id)
+    p = d / "status.json"
+    if not p.exists():
+        return None
+    with _lock:
+        entry = json.loads(p.read_text())
+        entry.update(kwargs)
+        entry["updated_at"] = time.time()
+        p.write_text(json.dumps(entry))
+    return entry
+
+
+def complete_request(request_id: str, result: dict):
+    entry = update_request(request_id, status="complete", success=True)
+    if entry:
+        (Path(DEFAULT_RESULTS_DIR) / request_id / "result.json").write_text(
+            json.dumps(result)
+        )
+
+
+def fail_request(request_id: str, error: str):
+    update_request(request_id, status="complete", success=False, error=error)
+
+
+def get_result(request_id: str) -> dict | None:
+    p = _req_path(request_id) / "result.json"
+    if not p.exists():
+        return None
+    return json.loads(p.read_text())
+
+
+def _cleanup_loop(interval: int = 300, ttl: int = DEFAULT_TTL):
+    while True:
+        time.sleep(interval)
+        now = time.time()
+        try:
+            results_dir = Path(DEFAULT_RESULTS_DIR)
+            if not results_dir.exists():
+                continue
+            for child in results_dir.iterdir():
+                if child.is_dir():
+                    status_p = child / "status.json"
+                    if status_p.exists():
+                        entry = json.loads(status_p.read_text())
+                        if now - entry.get("created_at", 0) > ttl:
+                            import shutil
+                            shutil.rmtree(str(child))
+        except Exception:
+            pass
+
+
+def start_cleanup_thread(interval: int = 300, ttl: int = DEFAULT_TTL):
+    t = threading.Thread(target=_cleanup_loop, args=(interval, ttl), daemon=True)
+    t.start()
+
+
+# ── Job queue for background processing (shared across workers via disk, but
+#    each worker has its own in-memory queue for picking up jobs) ──────
+_job_queue: list[tuple[str, str, bytes, str, dict]] = []
+_job_cond = threading.Condition()
+
+
+def submit_job(endpoint: str, raw_bytes: bytes, filename: str, opts: dict) -> str:
+    import uuid
+    request_id = str(uuid.uuid4())
+    create_request(request_id, {"endpoint": endpoint, "filename": filename})
+    with _job_cond:
+        _job_queue.append((request_id, endpoint, raw_bytes, filename, opts))
+        _job_cond.notify()
+    return request_id
+
+
+def wait_for_job(timeout: float | None = None) -> tuple | None:
+    with _job_cond:
+        if not _job_queue:
+            _job_cond.wait(timeout=timeout)
+        if not _job_queue:
+            return None
+        return _job_queue.pop(0)