diff --git a/src/lfx/src/lfx/_assets/component_index.json b/src/lfx/src/lfx/_assets/component_index.json index ebc4912823..1fbf57c0d5 100644 --- a/src/lfx/src/lfx/_assets/component_index.json +++ b/src/lfx/src/lfx/_assets/component_index.json @@ -66263,6 +66263,7 @@ "ignore_unsupported_extensions", "ignore_unspecified_files", "api_url", + "task_id", "max_concurrency", "max_poll_timeout", "api_headers", @@ -66272,7 +66273,7 @@ "icon": "Docling", "legacy": false, "metadata": { - "code_hash": "409d771a961e", + "code_hash": "5d06cbf9c92a", "dependencies": { "dependencies": [ { @@ -66318,28 +66319,47 @@ "template": { "_type": "Component", "api_headers": { - "_input_type": "NestedDictInput", + "_input_type": "TableInput", "advanced": true, "display_name": "HTTP headers", "dynamic": false, - "info": "Optional dictionary of additional headers required for connecting to Docling Serve.", + "info": "Optional headers required for connecting to Docling Serve.", "input_types": [ - "Message" + "Data", + "JSON" ], - "list": false, + "is_list": true, "list_add_label": "Add More", "name": "api_headers", "override_skip": false, "placeholder": "", + "real_time_refresh": true, "required": false, "show": true, + "table_icon": "Table", + "table_schema": [ + { + "description": "Key name", + "display_name": "Key", + "name": "key", + "type": "string" + }, + { + "description": "Value of the header", + "display_name": "Value", + "load_from_db": true, + "name": "value", + "type": "string" + } + ], "title_case": false, "tool_mode": false, - "trace_as_input": true, "trace_as_metadata": true, "track_in_telemetry": false, - "type": "NestedDict", - "value": {} + "trigger_icon": "Table", + "trigger_text": "Open table", + "type": "table", + "value": [] }, "api_url": { "_input_type": "StrInput", @@ -66378,7 +66398,7 @@ "show": true, "title_case": false, "type": "code", - "value": "import base64\nimport time\nfrom concurrent.futures import Future, ThreadPoolExecutor\nfrom pathlib import Path\nfrom typing import Any\n\nimport httpx\nfrom docling_core.types.doc import DoclingDocument\nfrom pydantic import ValidationError\n\nfrom lfx.base.data import BaseFileComponent\nfrom lfx.inputs import IntInput, NestedDictInput, StrInput\nfrom lfx.inputs.inputs import FloatInput\nfrom lfx.schema import Data\nfrom lfx.utils.util import transform_localhost_url\n\n\nclass DoclingRemoteComponent(BaseFileComponent):\n display_name = \"Docling Serve\"\n description = \"Uses Docling to process input documents connecting to your instance of Docling Serve.\"\n documentation = \"https://docling-project.github.io/docling/\"\n trace_type = \"tool\"\n icon = \"Docling\"\n name = \"DoclingRemote\"\n\n MAX_500_RETRIES = 5\n\n # https://docling-project.github.io/docling/usage/supported_formats/\n VALID_EXTENSIONS = [\n \"adoc\",\n \"asciidoc\",\n \"asc\",\n \"bmp\",\n \"csv\",\n \"dotx\",\n \"dotm\",\n \"docm\",\n \"docx\",\n \"htm\",\n \"html\",\n \"jpeg\",\n \"jpg\",\n \"json\",\n \"md\",\n \"pdf\",\n \"png\",\n \"potx\",\n \"ppsx\",\n \"pptm\",\n \"potm\",\n \"ppsm\",\n \"pptx\",\n \"tiff\",\n \"txt\",\n \"xls\",\n \"xlsx\",\n \"xhtml\",\n \"xml\",\n \"webp\",\n ]\n\n inputs = [\n *BaseFileComponent.get_base_inputs(),\n StrInput(\n name=\"api_url\",\n display_name=\"Server address\",\n info=\"URL of the Docling Serve instance.\",\n required=True,\n ),\n IntInput(\n name=\"max_concurrency\",\n display_name=\"Concurrency\",\n info=\"Maximum number of concurrent requests for the server.\",\n advanced=True,\n value=2,\n input_types=[\"Message\"],\n ),\n FloatInput(\n name=\"max_poll_timeout\",\n display_name=\"Maximum poll time\",\n info=\"Maximum waiting time for the document conversion to complete.\",\n advanced=True,\n value=3600,\n input_types=[\"Message\"],\n ),\n NestedDictInput(\n name=\"api_headers\",\n display_name=\"HTTP headers\",\n advanced=True,\n required=False,\n info=(\"Optional dictionary of additional headers required for connecting to Docling Serve.\"),\n input_types=[\"Message\"],\n ),\n NestedDictInput(\n name=\"docling_serve_opts\",\n display_name=\"Docling options\",\n advanced=True,\n required=False,\n info=(\n \"Optional dictionary of additional options. \"\n \"See https://github.com/docling-project/docling-serve/blob/main/docs/usage.md for more information.\"\n ),\n input_types=[\"Message\"],\n ),\n ]\n\n outputs = [\n *BaseFileComponent.get_base_outputs(),\n ]\n\n def process_files(self, file_list: list[BaseFileComponent.BaseFile]) -> list[BaseFileComponent.BaseFile]:\n # Transform localhost URLs to container-accessible hosts when running in a container\n transformed_url = transform_localhost_url(self.api_url)\n base_url = f\"{transformed_url}/v1\"\n\n def _convert_document(client: httpx.Client, file_path: Path, options: dict[str, Any]) -> Data | None:\n encoded_doc = base64.b64encode(file_path.read_bytes()).decode()\n payload = {\n \"options\": options,\n \"sources\": [{\"kind\": \"file\", \"base64_string\": encoded_doc, \"filename\": file_path.name}],\n }\n\n response = client.post(f\"{base_url}/convert/source/async\", json=payload)\n response.raise_for_status()\n task = response.json()\n\n http_failures = 0\n retry_status_start = 500\n retry_status_end = 600\n start_wait_time = time.monotonic()\n while task[\"task_status\"] not in (\"success\", \"failure\"):\n # Check if processing exceeds the maximum poll timeout\n processing_time = time.monotonic() - start_wait_time\n if processing_time >= self.max_poll_timeout:\n msg = (\n f\"Processing time {processing_time=} exceeds the maximum poll timeout {self.max_poll_timeout=}.\"\n \"Please increase the max_poll_timeout parameter or review why the processing \"\n \"takes long on the server.\"\n )\n self.log(msg)\n raise RuntimeError(msg)\n\n # Call for a new status update\n time.sleep(2)\n response = client.get(f\"{base_url}/status/poll/{task['task_id']}\")\n\n # Check if the status call gets into 5xx errors and retry\n if retry_status_start <= response.status_code < retry_status_end:\n http_failures += 1\n if http_failures > self.MAX_500_RETRIES:\n self.log(f\"The status requests got a http response {response.status_code} too many times.\")\n return None\n continue\n\n # Update task status\n task = response.json()\n\n result_resp = client.get(f\"{base_url}/result/{task['task_id']}\")\n result_resp.raise_for_status()\n result = result_resp.json()\n\n if \"json_content\" not in result[\"document\"] or result[\"document\"][\"json_content\"] is None:\n self.log(\"No JSON DoclingDocument found in the result.\")\n return None\n\n try:\n doc = DoclingDocument.model_validate(result[\"document\"][\"json_content\"])\n return Data(data={\"doc\": doc, \"file_path\": str(file_path)})\n except ValidationError as e:\n self.log(f\"Error validating the document. {e}\")\n return None\n\n docling_options = {\n \"to_formats\": [\"json\"],\n \"image_export_mode\": \"placeholder\",\n **(self.docling_serve_opts or {}),\n }\n\n processed_data: list[Data | None] = []\n with (\n httpx.Client(headers=self.api_headers) as client,\n ThreadPoolExecutor(max_workers=self.max_concurrency) as executor,\n ):\n futures: list[tuple[int, Future]] = []\n for i, file in enumerate(file_list):\n if file.path is None:\n processed_data.append(None)\n continue\n\n futures.append((i, executor.submit(_convert_document, client, file.path, docling_options)))\n\n for _index, future in futures:\n try:\n result_data = future.result()\n processed_data.append(result_data)\n except (httpx.HTTPStatusError, httpx.RequestError, KeyError, ValueError) as exc:\n self.log(f\"Docling remote processing failed: {exc}\")\n raise\n\n return self.rollup_data(file_list, processed_data)\n" + "value": "from __future__ import annotations\n\nimport base64\nimport json\nimport time\nfrom concurrent.futures import Future, ThreadPoolExecutor\nfrom pathlib import Path # noqa: TC003\nfrom typing import Any\n\nimport httpx\nfrom docling_core.types.doc import DoclingDocument\nfrom pydantic import ValidationError\n\nfrom lfx.base.data import BaseFileComponent\nfrom lfx.inputs import IntInput, NestedDictInput, StrInput, TableInput\nfrom lfx.inputs.inputs import FloatInput\nfrom lfx.schema import Data, dotdict\nfrom lfx.utils.util import transform_localhost_url\n\n\nclass DoclingRemoteComponent(BaseFileComponent):\n display_name = \"Docling Serve\"\n description = \"Uses Docling to process input documents connecting to your instance of Docling Serve.\"\n documentation = \"https://docling-project.github.io/docling/\"\n trace_type = \"tool\"\n icon = \"Docling\"\n name = \"DoclingRemote\"\n\n MAX_500_RETRIES = 5\n\n # https://docling-project.github.io/docling/usage/supported_formats/\n VALID_EXTENSIONS = [\n \"adoc\",\n \"asciidoc\",\n \"asc\",\n \"bmp\",\n \"csv\",\n \"dotx\",\n \"dotm\",\n \"docm\",\n \"docx\",\n \"htm\",\n \"html\",\n \"jpeg\",\n \"jpg\",\n \"json\",\n \"md\",\n \"pdf\",\n \"png\",\n \"potx\",\n \"ppsx\",\n \"pptm\",\n \"potm\",\n \"ppsm\",\n \"pptx\",\n \"tiff\",\n \"txt\",\n \"xls\",\n \"xlsx\",\n \"xhtml\",\n \"xml\",\n \"webp\",\n ]\n\n inputs = [\n *BaseFileComponent.get_base_inputs(),\n StrInput(\n name=\"api_url\",\n display_name=\"Server address\",\n info=\"URL of the Docling Serve instance.\",\n required=True,\n ),\n StrInput(\n name=\"task_id\",\n display_name=\"Task ID\",\n info=(\n \"Optional task ID from a previous Docling Serve upload. \"\n \"If provided, file input is ignored and the component polls for this task's results.\"\n ),\n required=False,\n ),\n IntInput(\n name=\"max_concurrency\",\n display_name=\"Concurrency\",\n info=\"Maximum number of concurrent requests for the server.\",\n advanced=True,\n value=2,\n input_types=[\"Message\"],\n ),\n FloatInput(\n name=\"max_poll_timeout\",\n display_name=\"Maximum poll time\",\n info=\"Maximum waiting time for the document conversion to complete.\",\n advanced=True,\n value=3600,\n input_types=[\"Message\"],\n ),\n TableInput(\n name=\"api_headers\",\n display_name=\"HTTP headers\",\n advanced=True,\n required=False,\n info=(\"Optional headers required for connecting to Docling Serve.\"),\n table_schema=[\n {\n \"name\": \"key\",\n \"display_name\": \"Key\",\n \"type\": \"string\",\n \"description\": \"Key name\",\n },\n {\n \"name\": \"value\",\n \"display_name\": \"Value\",\n \"load_from_db\": True,\n \"type\": \"string\",\n \"description\": \"Value of the header\",\n },\n ],\n value=[],\n real_time_refresh=True,\n input_types=[\"Data\", \"JSON\"],\n ),\n NestedDictInput(\n name=\"docling_serve_opts\",\n display_name=\"Docling options\",\n advanced=True,\n required=False,\n info=(\n \"Optional dictionary of additional options. \"\n \"See https://github.com/docling-project/docling-serve/blob/main/docs/usage.md for more information.\"\n ),\n input_types=[\"Message\"],\n ),\n ]\n\n outputs = [\n *BaseFileComponent.get_base_outputs(),\n ]\n\n @staticmethod\n def _add_header(headers: dict[str, str], key: Any, value: Any) -> None:\n key_str = str(key).strip()\n if not key_str or key_str == \"None\":\n return\n headers[key_str] = str(value)\n\n def _process_headers_input(self, headers_input: Any, component_headers_dict: dict[str, str]) -> None:\n if not headers_input:\n return\n\n items = headers_input if isinstance(headers_input, list) else [headers_input]\n\n for item in items:\n if not item:\n continue\n\n # Case 1: Data object\n if hasattr(item, \"data\") and isinstance(item.data, dict):\n data = item.data\n if \"key\" in data and \"value\" in data:\n self._add_header(component_headers_dict, data[\"key\"], data[\"value\"])\n else:\n # Fallback: merge all keys from Data object\n for k, v in data.items():\n if k not in (\"text_key\", \"default_value\"):\n self._add_header(component_headers_dict, k, v)\n\n # Case 2: Dictionary (Table row)\n elif isinstance(item, dict):\n if \"key\" in item and \"value\" in item:\n self._add_header(component_headers_dict, item[\"key\"], item[\"value\"])\n else:\n # Fallback: merge all keys\n for k, v in item.items():\n self._add_header(component_headers_dict, k, v)\n\n # Case 3: Message object\n elif hasattr(item, \"text\") and isinstance(item.text, str):\n try:\n parsed = json.loads(item.text)\n if isinstance(parsed, dict):\n for k, v in parsed.items():\n self._add_header(component_headers_dict, k, v)\n except json.JSONDecodeError:\n pass\n\n def _process_headers(self) -> dict[str, str]:\n \"\"\"Process the headers input into a valid dictionary.\"\"\"\n component_headers_dict: dict[str, str] = {}\n self._process_headers_input(self.api_headers, component_headers_dict)\n return component_headers_dict\n\n def update_build_config(self, build_config: dotdict, field_value: Any, field_name: str | None = None) -> dotdict:\n if field_name == \"api_headers\":\n if isinstance(field_value, dict):\n build_config[\"api_headers\"][\"value\"] = [{\"key\": k, \"value\": v} for k, v in field_value.items()]\n return build_config\n if field_value is None:\n build_config[\"api_headers\"][\"value\"] = []\n return build_config\n\n # Default behavior\n return super().update_build_config(build_config, field_value, field_name)\n\n def _poll_and_fetch_result(\n self, client: httpx.Client, base_url: str, task_id: str, file_path: str | None = None\n ) -> Data | None:\n \"\"\"Poll for task completion and fetch the result.\n\n Args:\n client: The HTTP client to use for requests.\n base_url: The base URL of the Docling Serve API.\n task_id: The task ID to poll for.\n file_path: Optional file path to include in the result data.\n\n Returns:\n Data object with the DoclingDocument, or None if processing failed.\n \"\"\"\n http_failures = 0\n retry_status_start = 500\n retry_status_end = 600\n start_wait_time = time.monotonic()\n\n task_status = None\n while task_status not in (\"success\", \"failure\"):\n processing_time = time.monotonic() - start_wait_time\n if processing_time >= self.max_poll_timeout:\n msg = (\n f\"Processing time {processing_time=} exceeds the maximum poll timeout {self.max_poll_timeout=}.\"\n \"Please increase the max_poll_timeout parameter or review why the processing \"\n \"takes long on the server.\"\n )\n self.log(msg)\n raise RuntimeError(msg)\n\n response = client.get(f\"{base_url}/status/poll/{task_id}\")\n\n if retry_status_start <= response.status_code < retry_status_end:\n http_failures += 1\n if http_failures > self.MAX_500_RETRIES:\n self.log(f\"The status requests got a http response {response.status_code} too many times.\")\n return None\n time.sleep(2)\n continue\n\n response.raise_for_status()\n task = response.json()\n task_status = task[\"task_status\"]\n if task_status not in (\"success\", \"failure\"):\n time.sleep(2)\n\n result_resp = client.get(f\"{base_url}/result/{task_id}\")\n result_resp.raise_for_status()\n result = result_resp.json()\n\n if result.get(\"status\") == \"failure\" or result.get(\"errors\"):\n errors = result.get(\"errors\", [])\n err_msg_list = []\n for err in errors:\n if isinstance(err, dict) and \"error_message\" in err:\n err_msg_list.append(err[\"error_message\"])\n elif isinstance(err, str):\n err_msg_list.append(err)\n\n err_details = \"; \".join(err_msg_list) if err_msg_list else \"Unknown Docling processing error\"\n\n msg = f\"Docling processing failed: {err_details}\"\n raise ValueError(msg)\n\n if \"json_content\" not in result[\"document\"] or result[\"document\"][\"json_content\"] is None:\n self.log(\"No JSON DoclingDocument found in the result.\")\n return None\n\n try:\n doc = DoclingDocument.model_validate(result[\"document\"][\"json_content\"])\n data_dict: dict[str, Any] = {\"doc\": doc}\n if file_path:\n data_dict[\"file_path\"] = file_path\n return Data(data=data_dict)\n except ValidationError as e:\n self.log(f\"Error validating the document. {e}\")\n return None\n\n def _process_task_id(self) -> list[Data]:\n \"\"\"Process an existing task by polling for status and retrieving results.\n\n Returns:\n List containing the result Data object, or empty list if processing failed.\n \"\"\"\n transformed_url = transform_localhost_url(self.api_url)\n base_url = f\"{transformed_url}/v1\"\n\n with httpx.Client(headers=self._process_headers()) as client:\n result = self._poll_and_fetch_result(client, base_url, self.task_id)\n return [result] if result else []\n\n def load_files_base(self) -> list[Data]:\n \"\"\"Load and process files, or poll an existing task if task_id is provided.\n\n Returns:\n list[Data]: Parsed data from the processed files or task.\n \"\"\"\n if self.task_id:\n return self._process_task_id()\n return super().load_files_base()\n\n def process_files(self, file_list: list[BaseFileComponent.BaseFile]) -> list[BaseFileComponent.BaseFile]:\n transformed_url = transform_localhost_url(self.api_url)\n base_url = f\"{transformed_url}/v1\"\n\n def _convert_document(client: httpx.Client, file_path: Path, options: dict[str, Any]) -> Data | None:\n encoded_doc = base64.b64encode(file_path.read_bytes()).decode()\n payload = {\n \"options\": options,\n \"sources\": [{\"kind\": \"file\", \"base64_string\": encoded_doc, \"filename\": file_path.name}],\n }\n\n response = client.post(f\"{base_url}/convert/source/async\", json=payload)\n response.raise_for_status()\n task = response.json()\n\n return self._poll_and_fetch_result(client, base_url, task[\"task_id\"], str(file_path))\n\n docling_options = {\n \"to_formats\": [\"json\"],\n \"image_export_mode\": \"placeholder\",\n **(self.docling_serve_opts or {}),\n }\n\n processed_data: list[Data | None] = []\n with (\n httpx.Client(headers=self._process_headers()) as client,\n ThreadPoolExecutor(max_workers=self.max_concurrency) as executor,\n ):\n futures: list[tuple[int, Future]] = []\n for i, file in enumerate(file_list):\n if file.path is None:\n processed_data.append(None)\n continue\n\n futures.append((i, executor.submit(_convert_document, client, file.path, docling_options)))\n\n for _index, future in futures:\n try:\n result_data = future.result()\n processed_data.append(result_data)\n except (httpx.HTTPStatusError, httpx.RequestError, KeyError, ValueError) as exc:\n self.log(f\"Docling remote processing failed: {exc}\")\n raise\n\n return self.rollup_data(file_list, processed_data)\n" }, "delete_server_file_after_processing": { "_input_type": "BoolInput", @@ -66633,6 +66653,27 @@ "track_in_telemetry": true, "type": "bool", "value": false + }, + "task_id": { + "_input_type": "StrInput", + "advanced": false, + "display_name": "Task ID", + "dynamic": false, + "info": "Optional task ID from a previous Docling Serve upload. If provided, file input is ignored and the component polls for this task's results.", + "list": false, + "list_add_label": "Add More", + "load_from_db": false, + "name": "task_id", + "override_skip": false, + "placeholder": "", + "required": false, + "show": true, + "title_case": false, + "tool_mode": false, + "trace_as_metadata": true, + "track_in_telemetry": false, + "type": "str", + "value": "" } }, "tool_mode": false @@ -120142,6 +120183,6 @@ "num_components": 360, "num_modules": 96 }, - "sha256": "335dd5ab0510deddde035bcd833035d786b874d243b1b93533838831ca5d4c42", + "sha256": "17d1b0468d27421f3c249c8fe417cf03c54f8747cdde47fc741f33707f5f8419", "version": "0.5.0" }