feat: Add RQ engine (#315)

Signed-off-by: Michele Dolfi <[email protected]>

Author: dolfim-ibm

.github/styles/config/vocabularies/Docling/accept.txt

@@ -5,6 +5,8 @@ async
 (?i)urls
 uvicorn
 [Ww]ebserver
+RQ
+(?i)url
 keyfile
 [Ww]ebsocket(s?)
 [Kk]ubernetes

docling_serve/__main__.py

@@ -11,6 +11,7 @@
 from rich.console import Console
 
 from docling_serve.settings import docling_serve_settings, uvicorn_settings
+from docling_serve.storage import get_scratch
 
 warnings.filterwarnings(action="ignore", category=UserWarning, module="pydantic|torch")
 warnings.filterwarnings(action="ignore", category=FutureWarning, module="easyocr")
@@ -361,6 +362,37 @@ def run(
     )
 
 
+@app.command()
+def rq_worker() -> Any:
+    """
+    Run the [bold]Docling JobKit[/bold] RQ worker.
+    """
+    from docling_jobkit.convert.manager import DoclingConverterManagerConfig
+    from docling_jobkit.orchestrators.rq.orchestrator import RQOrchestratorConfig
+    from docling_jobkit.orchestrators.rq.worker import run_worker
+
+    rq_config = RQOrchestratorConfig(
+        redis_url=docling_serve_settings.eng_rq_redis_url,
+        results_prefix=docling_serve_settings.eng_rq_results_prefix,
+        sub_channel=docling_serve_settings.eng_rq_sub_channel,
+        scratch_dir=get_scratch(),
+    )
+
+    cm_config = DoclingConverterManagerConfig(
+        artifacts_path=docling_serve_settings.artifacts_path,
+        options_cache_size=docling_serve_settings.options_cache_size,
+        enable_remote_services=docling_serve_settings.enable_remote_services,
+        allow_external_plugins=docling_serve_settings.allow_external_plugins,
+        max_num_pages=docling_serve_settings.max_num_pages,
+        max_file_size=docling_serve_settings.max_file_size,
+    )
+
+    run_worker(
+        rq_config=rq_config,
+        cm_config=cm_config,
+    )
+
+
 def main() -> None:
     app()
 

docling_serve/app.py

@@ -417,9 +417,17 @@ async def process_url(
                 detail=f"Conversion is taking too long. The maximum wait time is configure as DOCLING_SERVE_MAX_SYNC_WAIT={docling_serve_settings.max_sync_wait}.",
             )
 
-        task = await orchestrator.get_raw_task(task_id=task.task_id)
+        task_result = await orchestrator.task_result(task_id=task.task_id)
+        if task_result is None:
+            raise HTTPException(
+                status_code=404,
+                detail="Task result not found. Please wait for a completion status.",
+            )
         response = await prepare_response(
-            task=task, orchestrator=orchestrator, background_tasks=background_tasks
+            task_id=task.task_id,
+            task_result=task_result,
+            orchestrator=orchestrator,
+            background_tasks=background_tasks,
         )
         return response
 
@@ -457,9 +465,17 @@ async def process_file(
                 detail=f"Conversion is taking too long. The maximum wait time is configure as DOCLING_SERVE_MAX_SYNC_WAIT={docling_serve_settings.max_sync_wait}.",
             )
 
-        task = await orchestrator.get_raw_task(task_id=task.task_id)
+        task_result = await orchestrator.task_result(task_id=task.task_id)
+        if task_result is None:
+            raise HTTPException(
+                status_code=404,
+                detail="Task result not found. Please wait for a completion status.",
+            )
         response = await prepare_response(
-            task=task, orchestrator=orchestrator, background_tasks=background_tasks
+            task_id=task.task_id,
+            task_result=task_result,
+            orchestrator=orchestrator,
+            background_tasks=background_tasks,
         )
         return response
 
@@ -618,9 +634,17 @@ async def task_result(
         task_id: str,
     ):
         try:
-            task = await orchestrator.get_raw_task(task_id=task_id)
+            task_result = await orchestrator.task_result(task_id=task_id)
+            if task_result is None:
+                raise HTTPException(
+                    status_code=404,
+                    detail="Task result not found. Please wait for a completion status.",
+                )
             response = await prepare_response(
-                task=task, orchestrator=orchestrator, background_tasks=background_tasks
+                task_id=task_id,
+                task_result=task_result,
+                orchestrator=orchestrator,
+                background_tasks=background_tasks,
             )
             return response
         except TaskNotFoundError:

docling_serve/datamodel/responses.py

@@ -5,7 +5,7 @@
 
 from docling.datamodel.document import ConversionStatus, ErrorItem
 from docling.utils.profiling import ProfilingItem
-from docling_core.types.doc import DoclingDocument
+from docling_jobkit.datamodel.result import ExportDocumentResponse
 from docling_jobkit.datamodel.task_meta import TaskProcessingMeta
 
 
@@ -18,26 +18,19 @@ class ClearResponse(BaseModel):
     status: str = "ok"
 
 
-class DocumentResponse(BaseModel):
-    filename: str
-    md_content: Optional[str] = None
-    json_content: Optional[DoclingDocument] = None
-    html_content: Optional[str] = None
-    text_content: Optional[str] = None
-    doctags_content: Optional[str] = None
-
-
 class ConvertDocumentResponse(BaseModel):
-    document: DocumentResponse
+    document: ExportDocumentResponse
     status: ConversionStatus
     errors: list[ErrorItem] = []
     processing_time: float
     timings: dict[str, ProfilingItem] = {}
 
 
 class PresignedUrlConvertDocumentResponse(BaseModel):
-    status: ConversionStatus
     processing_time: float
+    num_converted: int
+    num_succeeded: int
+    num_failed: int
 
 
 class ConvertDocumentErrorResponse(BaseModel):

docling_serve/orchestrator_factory.py

@@ -3,6 +3,7 @@
 from docling_jobkit.orchestrators.base_orchestrator import BaseOrchestrator
 
 from docling_serve.settings import AsyncEngine, docling_serve_settings
+from docling_serve.storage import get_scratch
 
 
 @lru_cache
@@ -20,6 +21,7 @@ def get_async_orchestrator() -> BaseOrchestrator:
         local_config = LocalOrchestratorConfig(
             num_workers=docling_serve_settings.eng_loc_num_workers,
             shared_models=docling_serve_settings.eng_loc_share_models,
+            scratch_dir=get_scratch(),
         )
 
         cm_config = DoclingConverterManagerConfig(
@@ -33,6 +35,20 @@ def get_async_orchestrator() -> BaseOrchestrator:
         cm = DoclingConverterManager(config=cm_config)
 
         return LocalOrchestrator(config=local_config, converter_manager=cm)
+    elif docling_serve_settings.eng_kind == AsyncEngine.RQ:
+        from docling_jobkit.orchestrators.rq.orchestrator import (
+            RQOrchestrator,
+            RQOrchestratorConfig,
+        )
+
+        rq_config = RQOrchestratorConfig(
+            redis_url=docling_serve_settings.eng_rq_redis_url,
+            results_prefix=docling_serve_settings.eng_rq_results_prefix,
+            sub_channel=docling_serve_settings.eng_rq_sub_channel,
+            scratch_dir=get_scratch(),
+        )
+
+        return RQOrchestrator(config=rq_config)
     elif docling_serve_settings.eng_kind == AsyncEngine.KFP:
         from docling_jobkit.orchestrators.kfp.orchestrator import (
             KfpOrchestrator,