Repository: NVIDIA/nv-ingest

Issue:
[DOC]:  Skaffold doc page  non existent
### How would you describe the priority of this documentation request

None

### Please provide a link or source to the relevant docs

https://github.com/NVIDIA/nv-ingest/tree/main/skaffold

### Describe the problems in the documentation

I got this 404 error:

The main branch of  nv-ingest does not contain the path docs/kubernetes-dev.md.


when clicking on the link to Skaffold documentation in this page: https://github.com/NVIDIA/nv-ingest/tree/main/skaffold

### (Optional) Propose a correction or improvement

_No response_

Discussion:
copy-pr-bot[bot]: This pull request requires additional validation before any workflows can run on NVIDIA's runners.

Pull request vetters can view their responsibilities [here](https://docs.gha-runners.nvidia.com/cpr/vetters).

Contributors can view more details about this message [here](https://docs.gha-runners.nvidia.com/cpr/contributors).


Patch:
diff --git a/client/client_examples/examples/python_client_usage.ipynb b/client/client_examples/examples/python_client_usage.ipynb
index 867c5e2..0675089 100644
--- a/client/client_examples/examples/python_client_usage.ipynb
+++ b/client/client_examples/examples/python_client_usage.ipynb
@@ -9,11 +9,9 @@
     "\n",
     "This notebook provides a quick start guide to using the NV-Ingest Python API to create a client that interacts with a running NV-Ingest cluster. It will walk through the following:\n",
     "\n",
-    "- Instantiate a client object\n",
     "- Define the task configuration for an NV-Ingest job\n",
-    "- Submit a job the the NV-Ingest cluster\n",
-    "- Retrieve completed results\n",
-    "- Investigate the multimodal extractions\n"
+    "- Submit a job the the NV-Ingest cluster and retrieve completed results\n",
+    "- Investigate the multimodal extractions"
    ]
   },
   {
@@ -67,168 +65,70 @@
     "from base64 import b64decode\n",
     "import time\n",
     "\n",
-    "from nv_ingest_client.client import NvIngestClient\n",
-    "from nv_ingest_client.message_clients.rest.rest_client import RestClient\n",
-    "from nv_ingest_client.primitives import JobSpec\n",
-    "from nv_ingest_client.primitives.tasks import DedupTask\n",
-    "from nv_ingest_client.primitives.tasks import EmbedTask\n",
-    "from nv_ingest_client.primitives.tasks import ExtractTask\n",
-    "from nv_ingest_client.primitives.tasks import FilterTask\n",
-    "from nv_ingest_client.primitives.tasks import SplitTask\n",
-    "from nv_ingest_client.primitives.tasks import StoreTask, StoreEmbedTask\n",
-    "from nv_ingest_client.primitives.tasks import VdbUploadTask\n",
-    "from nv_ingest_client.util.file_processing.extract import extract_file_content\n",
+    "from nv_ingest_client.client import Ingestor\n",
     "\n",
     "from IPython import display"
    ]
   },
   {
    "cell_type": "markdown",
-   "id": "6bc54af4",
+   "id": "69045f8a",
    "metadata": {},
    "source": [
-    "Load a sample PDF to demonstrate NV-Ingest usage."
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "16298e5f",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "file_content, file_type = extract_file_content(SAMPLE_PDF)"
-   ]
-  },
-  {
-   "cell_type": "markdown",
-   "id": "61306c09",
-   "metadata": {},
-   "source": [
-    "Initialize a client that will submit jobs to our NV-Ingest cluster."
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "1d31e44c",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "client = NvIngestClient(\n",
-    "    message_client_allocator=RestClient,\n",
-    "    message_client_hostname=HTTP_HOST,\n",
-    "    message_client_port=HTTP_PORT,\n",
-    "    message_client_kwargs=None,\n",
-    "    msg_counter_id=\"nv-ingest-message-id\",\n",
-    "    worker_pool_size=1,\n",
-    ")"
-   ]
-  },
-  {
-   "cell_type": "markdown",
-   "id": "92211ced",
-   "metadata": {},
-   "source": [
-    "A `JobSpec` is a specification for creating a job for submission to the NV-Ingest microservice. It accepts the following parameters:\n",
+    "Each ingest job includes a set of tasks. These tasks define the operations that will be performed during ingestion. This allows each job to potentially have different ingestion instructions. Here we define a simple extract oriented job, but the full list of supported options are contained below:\n",
     "\n",
-    "- `document_type` : The file type of the file to be ingested.\n",
-    "- `payload` : A base64 encoded string of the file to be ingested.\n",
-    "- `source_id` : An identifier that maps to the file, our example uses the filename here.\n",
-    "- `source_name` : The name of the source for this ingest job, again, we use the filename.\n",
-    "- `extented_options` : A dictionary of additional options to pass in to the ingest job, we pass in informations allowing us to conduct performance tracing of the job.\n"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "d00dd717",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "job_spec = JobSpec(\n",
-    "    document_type=file_type,\n",
-    "    payload=file_content,\n",
-    "    source_id=SAMPLE_PDF,\n",
-    "    source_name=SAMPLE_PDF,\n",
-    "    extended_options={\n",
-    "        \"tracing_options\": {\n",
-    "            \"trace\": True,\n",
-    "            \"ts_send\": time.time_ns(),\n",
-    "        }\n",
-    "    },\n",
-    ")"
-   ]
-  },
-  {
-   "cell_type": "markdown",
-   "id": "6a0a87cb",
-   "metadata": {},
-   "source": [
-    "Each ingest job will include a set of tasks. These tasks will define the operations that will be performed during ingestion. This allows each job to potentially have different ingestion instructions. Here we define a simple extract oriented job, but the full list of supported options are contained below:\n",
+    "- `extract` : Performs multimodal extractions from a document, including text, images, and tables.\n",
+    "- `split` : Chunk the text into smaller chunks, useful for storing in a vector database for retrieval applications.\n",
+    "- `dedup` : Identifies duplicate images in document that can be filtered to remove data redundancy.\n",
+    "- `filter` : Filters out images that are likely not useful using some heuristics, including size and aspect ratio.\n",
+    "- `embed` : Computes an embedding for the extracted content using a [`nvidia/nv-embedqa-e5-v5`](https://catalog.ngc.nvidia.com/orgs/nim/teams/nvidia/containers/nv-embedqa-e5-v5) NVIDIA Inference Microservice (NIM) by default.\n",
+    "- `store` : Save the extracted tables or images to an S3 compliant object store like MinIO.\n",
+    "- `vbd_upload` : Save embeddings, chunks, and metadata to a Milvus vector database.\n",
     "\n",
-    "- `ExtractTask` : Performs multimodal extractions from a document, including text, images, and tables.\n",
-    "- `TableExtractTask`: Extracts the content from tables.\n",
-    "- `ChartExtractTask`: Extracts the content from charts.\n",
-    "- `SplitTask` : Chunk the text into smaller chunks, useful for storing in a vector database for retrieval applications.\n",
-    "- `Dedup` : Identifies duplicate images in document that can be filtered to remove data redundancy.\n",
-    "- `Filter` : Filters out images that are likely not useful using some heuristics, including size and aspect ratio.\n",
-    "- `EmbedTask` : Pass the text or table extractions through `\"nvidia/nv-embedqa-e5-v5` NIM to obtain its embeddings.\n",
-    "- `Store` : Save the extracted tables or images to an S3 compliant object store like MinIO.\n",
-    "- `Upload` : Save embeddings, chunks, and metadata to a Milvus vector database.\n",
-    "\n",
-    "After defining the ingestion tasks, they must be added to the job specification."
+    "We'll use the Ingestor interface to chain together an extraction tast and a deduplication task to ingest our sample PDF. "
    ]
   },
   {
    "cell_type": "code",
    "execution_count": null,
-   "id": "061e5b4d",
+   "id": "e1f654ca",
    "metadata": {},
    "outputs": [],
    "source": [
-    "extract_task = ExtractTask(\n",
-    "    document_type=file_type,\n",
-    "    extract_text=True,\n",
-    "    extract_images=True,\n",
-    "    extract_tables=True,\n",
-    "    text_depth=\"document\",\n",
-    "    extract_tables_method=\"yolox\",\n",
-    ")\n",
-    "\n",
-    "dedup_task = DedupTask(\n",
-    "    content_type=\"image\",\n",
-    "    filter=True,\n",
-    ")\n",
-    "\n",
-    "job_spec.add_task(extract_task)\n",
-    "job_spec.add_task(dedup_task)"
+    "SAMPLE_PDF = \"../../../data/multimodal_test.pdf\"\n",
+    "\n",
+    "ingestor = (\n",
+    "    Ingestor(message_client_hostname=HTTP_HOST)\n",
+    "    .files(SAMPLE_PDF)\n",
+    "    .extract(\n",
+    "        extract_text=True,\n",
+    "        extract_tables=True,\n",
+    "        extract_charts=True,\n",
+    "        extract_images=True,\n",
+    "        text_depth=\"document\",\n",
+    "    ).dedup(\n",
+    "        content_type=\"image\",\n",
+    "        filter=True,\n",
+    "    )\n",
+    ")"
    ]
   },
   {
    "cell_type": "markdown",
-   "id": "2da2f0ff",
+   "id": "1875df4a",
    "metadata": {},
    "source": [
-    "A job identifier is created for the job specification. This is used to retrieve the results upon completion.\n",
-    "\n",
-    "With the `job_id`, the job is submitted to the NV-Ingest microservice. When the job is complete, the results are fetched.\n",
-    "\n",
-    "Note, many jobs can be submitted and asynchronous execution is supported."
+    "Submit the job to our NV-Ingest cluster"
    ]
   },
   {
    "cell_type": "code",
    "execution_count": null,
-   "id": "c9ad0589",
+   "id": "119c7db6",
    "metadata": {},
    "outputs": [],
    "source": [
-    "job_id = client.add_job(job_spec)\n",
-    "client.submit_job(job_id, TASK_QUEUE)\n",
-    "generated_metadata = client.fetch_job_result(\n",
-    "    job_id, timeout=DEFAULT_JOB_TIMEOUT\n",
-    ")[0]"
+    "generated_metadata = ingestor.ingest()[0]"
    ]
   },
   {
@@ -524,9 +424,11 @@
    "source": [
     "This section illustrates usage of the remaining task types used when supporting retrieval workflows.\n",
     "\n",
-    "- `StoreTask` - Stores extracted content to an S3 compliant object store (MinIO by default) and updates the `source_metadata` with the corresponding stored location.\n",
-    "- `EmbedTask` - Computes an embedding for the extracted content using a [`nvidia/nv-embedqa-e5-v5`](https://catalog.ngc.nvidia.com/orgs/nim/teams/nvidia/containers/nv-embedqa-e5-v5) NVIDIA Inference Microservice (NIM) by default.\n",
-    "- `VdbUploadTask` - Inserts ingested content into a Milvus vector database to support retrieval use cases."
+    "- `filter` : Filters out images that are likely not useful using some heuristics, including size and aspect ratio.\n",
+    "- `split` : Chunk the text into smaller chunks, useful for storing in a vector database for retrieval applications.\n",
+    "- `store` - Stores extracted content to an S3 compliant object store (MinIO by default) and updates the `source_metadata` with the corresponding stored location.\n",
+    "- `embed` - Computes an embedding for the extracted content using a [`nvidia/nv-embedqa-e5-v5`](https://catalog.ngc.nvidia.com/orgs/nim/teams/nvidia/containers/nv-embedqa-e5-v5) NVIDIA Inference Microservice (NIM) by default.\n",
+    "- `vdb_upload` - Inserts ingested content into a Milvus vector database to support retrieval use cases."
    ]
   },
   {
@@ -534,145 +436,106 @@
    "id": "856422c5",
    "metadata": {},
    "source": [
-    "Define the ingest job specification."
+    "Define the ingest job specification. Here the task configuration is expanded, but requires the ancillary services (Embedding NIM, MinIO object stor, and Milvus Vector Database) to be up and running to return metadata back to the client."
    ]
   },
   {
    "cell_type": "code",
    "execution_count": null,
-   "id": "97c8cd73",
+   "id": "110b7467",
    "metadata": {},
    "outputs": [],
    "source": [
-    "job_spec = JobSpec(\n",
-    "    document_type=file_type,\n",
-    "    payload=file_content,\n",
-    "    source_id=SAMPLE_PDF,\n",
-    "    source_name=SAMPLE_PDF,\n",
-    "    extended_options={\n",
-    "        \"tracing_options\": {\n",
-    "            \"trace\": True,\n",
-    "            \"ts_send\": time.time_ns(),\n",
+    "ingestor = (\n",
+    "    Ingestor(message_client_hostname=HTTP_HOST)\n",
+    "    .files(SAMPLE_PDF)\n",
+    "    .extract(\n",
+    "        extract_text=True,\n",
+    "        extract_tables=True,\n",
+    "        extract_charts=True,\n",
+    "        extract_images=True,\n",
+    "        text_depth=\"document\",\n",
+    "    ).dedup(\n",
+    "        content_type=\"image\",\n",
+    "        filter=True,\n",
+    "    ).filter(\n",
+    "        content_type=\"image\",\n",
+    "        min_size=128,\n",
+    "        max_aspect_ratio=5.0,\n",
+    "        min_aspect_ratio=0.2,\n",
+    "        filter=True,\n",
+    "    ).split(\n",
+    "        split_by=\"word\",\n",
+    "        split_length=300,\n",
+    "        split_overlap=10,\n",
+    "        max_character_length=5000,\n",
+    "        sentence_window_size=0,\n",
+    "    ).store(\n",
+    "        structured=True,\n",
+    "        images=True,\n",
+    "        store_method=\"minio\",\n",
+    "        params={\n",
+    "            \"access_key\": MINIO_ACCESS_KEY, \n",
+    "            \"secret_key\": MINIO_SECRET_KEY,\n",
     "        }\n",
-    "    },\n",
+    "    )\n",
+    "    .embed()\n",
+    "    .vdb_upload()\n",
     ")"
    ]
   },
   {
    "cell_type": "markdown",
-   "id": "3e4bab16",
+   "id": "82d1db56",
    "metadata": {},
    "source": [
-    "Here the task configuration is expanded, but requires the ancillary services (Embedding NIM, MinIO object stor, and Milvus Vector Database) to be up and running to return metadata back to the client."
+    "Submit the job and retrieve the results"
    ]
   },
   {
    "cell_type": "code",
    "execution_count": null,
-   "id": "052ef358",
+   "id": "e30e87de",
    "metadata": {},
    "outputs": [],
    "source": [
-    "extract_task = ExtractTask(\n",
-    "    document_type=file_type,\n",
-    "    extract_text=True,\n",
-    "    extract_images=True,\n",
-    "    extract_tables=True,\n",
-    "    text_depth=\"document\",\n",
-    "    extract_tables_method=\"yolox\",\n",
-    ")\n",
-    "\n",
-    "dedup_task = DedupTask(\n",
-    "    content_type=\"image\",\n",
-    "    filter=True,\n",
-    ")\n",
-    "\n",
-    "filter_task = FilterTask(\n",
-    "    content_type=\"image\",\n",
-    "    min_size=128,\n",
-    "    max_aspect_ratio=5.0,\n",
-    "    min_aspect_ratio=0.2,\n",
-    "    filter=True,\n",
-    ")\n",
-    "\n",
-    "split_task = SplitTask(\n",
-    "    split_by=\"word\",\n",
-    "    split_length=300,\n",
-    "    split_overlap=10,\n",
-    "    max_character_length=5000,\n",
-    "    sentence_window_size=0,\n",
-    ")\n",
-    "\n",
-    "store_task = StoreTask(\n",
-    "    structured=True,\n",
-    "    images=True,\n",
-    "    store_method=\"minio\",\n",
-    "    params={\n",
-    "        \"access_key\": MINIO_ACCESS_KEY, \n",
-    "        \"secret_key\": MINIO_SECRET_KEY,\n",
-    "    }\n",
-    ")\n",
-    "\n",
-    "embed_task = EmbedTask(\n",
-    "    text=True,\n",
-    "    tables=True,\n",
-    ")\n",
-    "\n",
-    "store_embed_task = StoreEmbedTask(\n",
-    "    params={\n",
-    "        \"access_key\": MINIO_ACCESS_KEY, \n",
-    "        \"secret_key\": MINIO_SECRET_KEY,\n",
-    "    }\n",
-    "\n",
-    ")\n",
-    "\n",
-    "vdb_upload_task = VdbUploadTask(\n",
-    "    bulk_ingest=True,\n",
-    "    bulk_ingest_path=\"embeddings/\",\n",
-    "    params={\n",
-    "        \"access_key\": MINIO_ACCESS_KEY, \n",
-    "        \"secret_key\": MINIO_SECRET_KEY,\n",
-    "    }\n",
-    ")\n",
-    "\n",
-    "\n",
-    "job_spec.add_task(extract_task)\n",
-    "job_spec.add_task(dedup_task)\n",
-    "job_spec.add_task(filter_task)\n",
-    "job_spec.add_task(split_task)\n",
-    "job_spec.add_task(store_task)\n",
-    "job_spec.add_task(embed_task)\n",
-    "job_spec.add_task(vdb_upload_task)"
+    "generated_metadata = ingestor.ingest()[0]"
    ]
   },
   {
    "cell_type": "markdown",
-   "id": "deca53ee",
+   "id": "eb8999e2-acbd-48e6-a30f-4cea5e745152",
    "metadata": {},
    "source": [
-    "A job identifier is created for the job specification. This is used to retrieve the results upon completion.\n",
-    "\n",
-    "With the `job_id`, the job is submitted to the NV-Ingest cluster. When the job is complete, the results are fetched."
+    "Query the Milvus VDB"
    ]
   },
   {
    "cell_type": "code",
    "execution_count": null,
-   "id": "d471133b",
+   "id": "17d2ea9a-dc2e-4570-bde9-5a5317a66ce9",
    "metadata": {},
    "outputs": [],
    "source": [
-    "job_id = client.add_job(job_spec)\n",
-    "client.submit_job(job_id, TASK_QUEUE)\n",
-    "generated_metadata = client.fetch_job_result(\n",
-    "    job_id, timeout=DEFAULT_JOB_TIMEOUT\n",
-    ")[0][0]"
+    "from nv_ingest_client.util.milvus import nvingest_retrieval\n",
+    "\n",
+    "query = \"What is the dog doing and where?\"\n",
+    "\n",
+    "nvingest_retrieval(\n",
+    "        [query],\n",
+    "        \"nv_ingest_collection\",\n",
+    "        hybrid=False,\n",
+    "        embedding_endpoint=\"http://localhost:8012/v1\",\n",
+    "        model_name=\"nvidia/nv-embedqa-e5-v5\",\n",
+    "        top_k=1,\n",
+    "        gpu_search=True,\n",
+    ")"
    ]
   },
   {
    "cell_type": "code",
    "execution_count": null,
-   "id": "c3386304-e3a8-40a7-a09a-34a72059404f",
+   "id": "eada27cb-d7b6-4e47-a09a-483ea9d758fa",
    "metadata": {},
    "outputs": [],
    "source": []
diff --git a/docs/docs/user-guide/developer-guide/telemetry.md b/docs/docs/user-guide/developer-guide/telemetry.md
index f3f6df1..9732549 100644
--- a/docs/docs/user-guide/developer-guide/telemetry.md
+++ b/docs/docs/user-guide/developer-guide/telemetry.md
@@ -11,7 +11,7 @@ $ docker compose up otel-collector
 
 Once OpenTelemetry and Zipkin are running, you can open your browser to explore traces: http://$YOUR_DOCKER_HOST:9411/zipkin/.
 
-![](images/zipkin.png)
+![](../../assets/images/zipkin.png)
 
 To run Prometheus, run:
 
@@ -21,4 +21,4 @@ $ docker compose up prometheus
 
 Once Promethus is running, you can open your browser to explore metrics: [http://$YOUR_DOCKER_HOST:9090/]
 
-![](images/prometheus.png)
+![](../../assets/images/prometheus.png)
diff --git a/skaffold/README.md b/skaffold/README.md
index ee1aa37..02d57ec 100644
--- a/skaffold/README.md
+++ b/skaffold/README.md
@@ -2,6 +2,6 @@
 
 Skaffold is intended to support the nv-ingest development team with kubernetes dev and testing. It is not meant to be used in production deployments or even for local testing.
 
-We offer k8s support via Helm and those instructions can be found at [Helm Documentation](../helm/README.md)
+We offer k8s support via Helm and those instructions can be found at [Helm Documentation](../helm/README.md).
 
-For developers further documentation for using Skaffold can be found at [Skaffold Documentation](../docs/kubernetes-dev.md)
+For developers further documentation for using Skaffold can be found at [Skaffold Documentation](/docs/docs/user-guide/developer-guide/kubernetes-dev.md).
diff --git a/src/nv_ingest/stages/multiprocessing_stage.py b/src/nv_ingest/stages/multiprocessing_stage.py
index a86a506..94d46e7 100644
--- a/src/nv_ingest/stages/multiprocessing_stage.py
+++ b/src/nv_ingest/stages/multiprocessing_stage.py
@@ -489,6 +489,8 @@ class MultiProcessingBaseStage(SinglePortStage):
                 # This is the first location where we have access to both the control message and the work package,
                 # if we had any errors in the processing, raise them here.
                 if work_package.get("error", False):
+                    logger.error(f"Error in processing: {work_package['error_message']}")
+
                     raise RuntimeError(work_package["error_message"])
 
                 gdf = cudf.from_pandas(work_package["payload"])
diff --git a/src/nv_ingest/stages/nim/table_extraction.py b/src/nv_ingest/stages/nim/table_extraction.py
index 126909f..8e77312 100644
--- a/src/nv_ingest/stages/nim/table_extraction.py
+++ b/src/nv_ingest/stages/nim/table_extraction.py
@@ -94,6 +94,7 @@ def _update_metadata(
                 except Exception as e:
                     logger.error(f"Error processing image {i}. Error: {e}", exc_info=True)
                     results[i] = (b64_image, ("", ""))
+                    raise
 
     # ------------------------------------------------
     # HTTP path: submit requests in batches.
diff --git a/src/nv_ingest/util/nim/cached.py b/src/nv_ingest/util/nim/cached.py
index d424ced..bed6976 100644
--- a/src/nv_ingest/util/nim/cached.py
+++ b/src/nv_ingest/util/nim/cached.py
@@ -7,7 +7,7 @@ import base64
 import io
 import logging
 import PIL.Image as Image
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Optional, List
 
 import numpy as np
 
@@ -19,8 +19,8 @@ logger = logging.getLogger(__name__)
 
 class CachedModelInterface(ModelInterface):
     """
-    An interface for handling inference with a Cached model, supporting both gRPC and HTTP protocols,
-    including batched input.
+    An interface for handling inference with a Cached model, supporting both gRPC and HTTP
+    protocols, including batched input.
     """
 
     def name(self) -> str:
@@ -36,95 +36,101 @@ class CachedModelInterface(ModelInterface):
 
     def prepare_data_for_inference(self, data: Dict[str, Any]) -> Dict[str, Any]:
         """
-        Prepare input data for inference by decoding base64 images into numpy arrays.
+        Decode base64-encoded images into NumPy arrays, storing them in `data["image_arrays"]`.
 
         Parameters
         ----------
-        data : dict
-            The input data containing either a single "base64_image" or multiple "base64_images".
+        data : dict of str -> Any
+            The input data containing either:
+             - "base64_image": a single base64-encoded image, or
+             - "base64_images": a list of base64-encoded images.
 
         Returns
         -------
-        dict
-            The updated data dictionary with decoded image arrays stored in "image_arrays".
+        dict of str -> Any
+            The updated data dictionary with decoded image arrays stored in
+            "image_arrays", where each array has shape (H, W, C).
+
+        Raises
+        ------
+        KeyError
+            If neither 'base64_image' nor 'base64_images' is provided.
+        ValueError
+            If 'base64_images' is provided but is not a list.
         """
-        # Handle single vs. multiple images.
-        # For batch processing, prefer "base64_images".
         if "base64_images" in data:
             base64_list = data["base64_images"]
             if not isinstance(base64_list, list):
                 raise ValueError("The 'base64_images' key must contain a list of base64-encoded strings.")
             data["image_arrays"] = [base64_to_numpy(img) for img in base64_list]
+
         elif "base64_image" in data:
-            # Fallback to single image case; wrap it in a list to keep everything consistent
+            # Fallback to single image case; wrap it in a list to keep the interface consistent
             data["image_arrays"] = [base64_to_numpy(data["base64_image"])]
+
         else:
-            raise KeyError(
-                "Input data must include either 'base64_image' or 'base64_images' with base64-encoded images."
-            )
+            raise KeyError("Input data must include 'base64_image' or 'base64_images' with base64-encoded images.")
 
         return data
 
     def format_input(self, data: Dict[str, Any], protocol: str) -> Any:
         """
-        Format input data for the specified protocol (gRPC or HTTP).
+        Format input data for the specified protocol ("grpc" or "http"), handling batched images.
 
         Parameters
         ----------
-        data : dict
-            The input data to format, containing "image_arrays" (list of np.ndarray).
+        data : dict of str -> Any
+            The input data dictionary, expected to contain "image_arrays" (a list of np.ndarray).
         protocol : str
-            The protocol to use ("grpc" or "http").
+            The protocol to use, "grpc" or "http".
 
         Returns
         -------
         Any
-            The formatted input data.
+            The formatted input data. For gRPC, a single NumPy array of shape (B, H, W, C).
+            For HTTP, a JSON-serializable dict containing base64-encoded images.
 
         Raises
         ------
+        KeyError
+            If "image_arrays" is missing in the data dictionary.
         ValueError
-            If an invalid protocol is specified or if the images do not share the same shape.
+            If the protocol is invalid, or if images have differing shapes for gRPC.
         """
         if "image_arrays" not in data:
             raise KeyError("Expected 'image_arrays' in data. Make sure prepare_data_for_inference was called.")
 
-        # The arrays we got from prepare_data_for_inference
         image_arrays = data["image_arrays"]
 
         if protocol == "grpc":
             logger.debug("Formatting input for gRPC Cached model (batched).")
 
-            batched_images = []
+            batched_images: List[np.ndarray] = []
             for arr in image_arrays:
-                # If shape is (H, W, C), expand to (1, H, W, C)
-                # If already has a leading batch dimension, keep it
+                # Expand from (H, W, C) to (1, H, W, C) if needed
                 if arr.ndim == 3:
-                    arr = np.expand_dims(arr, axis=0)  # -> (1, H, W, C)
-
+                    arr = np.expand_dims(arr, axis=0)
+                # Convert to float32
                 batched_images.append(arr.astype(np.float32))
 
             if not batched_images:
                 raise ValueError("No valid images found for gRPC formatting.")
 
-            # Check that all images match in shape beyond the batch dimension
-            # e.g. every array should be (1, H, W, C) with the same (H, W, C)
-            shapes = [img.shape[1:] for img in batched_images]  # list of (H, W, C) shapes
+            # Ensure all images have the same shape (beyond batch dimension)
+            shapes = [img.shape[1:] for img in batched_images]  # each is (H, W, C)
             if any(s != shapes[0] for s in shapes[1:]):
                 raise ValueError(f"All images must have the same dimensions for gRPC batching. Found: {shapes}")
 
-            # Concatenate along the batch dimension => shape (B, H, W, C)
+            # Concatenate => shape (B, H, W, C)
             batched_input = np.concatenate(batched_images, axis=0)
             return batched_input
 
         elif protocol == "http":
             logger.debug("Formatting input for HTTP Cached model (batched).")
-            # Convert each image back to base64, building a single payload with multiple images
-            # to mimic YOLOX or NIM's typical batch approach
 
-            content_list = []
+            content_list: List[Dict[str, Any]] = []
             for arr in image_arrays:
-                # Convert from np.uint8 or float -> PIL -> base64
+                # Convert to uint8 if needed, then to PIL, then to base64
                 if arr.dtype != np.uint8:
                     arr = (arr * 255).astype(np.uint8)
                 image_pil = Image.fromarray(arr)
@@ -132,7 +138,7 @@ class CachedModelInterface(ModelInterface):
                 image_pil.save(buffered, format="PNG")
                 base64_img = base64.b64encode(buffered.getvalue()).decode("utf-8")
 
-                # Build item for Nim
+                # Build item for Nim-like structure
                 image_item = {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{base64_img}"}}
                 content_list.append(image_item)
 
@@ -144,32 +150,37 @@ class CachedModelInterface(ModelInterface):
         else:
             raise ValueError("Invalid protocol specified. Must be 'grpc' or 'http'.")
 
-    def parse_output(self, response: Any, protocol: str, data: Optional[Dict[str, Any]] = None, **kwargs) -> Any:
+    def parse_output(self, response: Any, protocol: str, data: Optional[Dict[str, Any]] = None, **kwargs: Any) -> Any:
         """
-        Parse the output from the model's inference response.
+        Parse the output from the Cached model's inference response.
 
         Parameters
         ----------
         response : Any
-            The response from the model inference.
+            The raw response from the model inference.
         protocol : str
             The protocol used ("grpc" or "http").
-        data : dict, optional
-            Additional input data passed to the function (could contain 'image_arrays').
+        data : dict of str -> Any, optional
+            Additional input data (unused here, but available for consistency).
+        **kwargs : Any
+            Additional keyword arguments for future compatibility.
 
         Returns
         -------
         Any
-            The parsed output data (likely a list of strings or a single string).
+            The parsed output data (e.g., list of strings), depending on the protocol.
 
         Raises
         ------
         ValueError
-            If an invalid protocol is specified.
+            If the protocol is invalid.
+        RuntimeError
+            If the HTTP response is not as expected (missing 'data' key).
         """
         if protocol == "grpc":
             logger.debug("Parsing output from gRPC Cached model (batched).")
-            parsed = []
+            parsed: List[str] = []
+            # Assume `response` is iterable, each element a list/array of byte strings
             for single_output in response:
                 joined_str = " ".join(o.decode("utf-8") for o in single_output)
                 parsed.append(joined_str)
@@ -179,21 +190,21 @@ class CachedModelInterface(ModelInterface):
             logger.debug("Parsing output from HTTP Cached model (batched).")
             if not isinstance(response, dict):
                 raise RuntimeError("Expected JSON/dict response for HTTP, got something else.")
-
             if "data" not in response or not response["data"]:
                 raise RuntimeError("Unexpected response format: 'data' key missing or empty.")
 
-            contents = []
+            contents: List[str] = []
             for item in response["data"]:
                 # Each "item" might have a "content" key
                 content = item.get("content", "")
                 contents.append(content)
 
             return contents
+
         else:
             raise ValueError("Invalid protocol specified. Must be 'grpc' or 'http'.")
 
-    def process_inference_results(self, output: Any, protocol: str, **kwargs) -> Any:
+    def process_inference_results(self, output: Any, protocol: str, **kwargs: Any) -> Any:
         """
         Process inference results for the Cached model.
 
@@ -201,13 +212,17 @@ class CachedModelInterface(ModelInterface):
         ----------
         output : Any
             The raw output from the model.
+        protocol : str
+            The inference protocol used ("grpc" or "http").
+        **kwargs : Any
+            Additional parameters for post-processing (not used here).
 
         Returns
         -------
         Any
-            The processed inference results.
+            The processed inference results, which here is simply returned as-is.
         """
-        # For Cached model, the output is the chart content as a string
+        # For Cached model, we simply return what we parsed (e.g., a list of strings or a single string)
         return output
 
     def _extract_content_from_nim_response(self, json_response: Dict[str, Any]) -> Any:
@@ -216,7 +231,7 @@ class CachedModelInterface(ModelInterface):
 
         Parameters
         ----------
-        json_response : dict
+        json_response : dict of str -> Any
             The JSON response from the NIM API.
 
         Returns
@@ -227,7 +242,7 @@ class CachedModelInterface(ModelInterface):
         Raises
         ------
         RuntimeError
-            If the response does not contain the expected "data" key or if it is empty.
+            If the response format is unexpected (missing 'data' or empty).
         """
         if "data" not in json_response or not json_response["data"]:
             raise RuntimeError("Unexpected response format: 'data' key is missing or empty.")
diff --git a/src/nv_ingest/util/nim/helpers.py b/src/nv_ingest/util/nim/helpers.py
index 9ee30f9..9e6e40d 100644
--- a/src/nv_ingest/util/nim/helpers.py
+++ b/src/nv_ingest/util/nim/helpers.py
@@ -185,31 +185,38 @@ class NimClient:
             If an invalid protocol is specified.
         """
 
-        # Prepare data for inference
-        prepared_data = self.model_interface.prepare_data_for_inference(data)
+        try:
+            # Prepare data for inference
+            prepared_data = self.model_interface.prepare_data_for_inference(data)
+
+            # Format input based on protocol
+            formatted_input = self.model_interface.format_input(prepared_data, protocol=self.protocol)
+
+            # Perform inference
+            if self.protocol == "grpc":
+                logger.debug("Performing gRPC inference...")
+                response = self._grpc_infer(formatted_input, model_name)
+                logger.debug("gRPC inference received response")
+            elif self.protocol == "http":
+                logger.debug("Performing HTTP inference...")
+                response = self._http_infer(formatted_input)
+                logger.debug("HTTP inference received response")
+            else:
+                raise ValueError("Invalid protocol specified. Must be 'grpc' or 'http'.")
 
-        # Format input based on protocol
-        formatted_input = self.model_interface.format_input(prepared_data, protocol=self.protocol)
+            # Parse and process output
+            parsed_output = self.model_interface.parse_output(
+                response, protocol=self.protocol, data=prepared_data, **kwargs
+            )
+            results = self.model_interface.process_inference_results(
+                parsed_output, original_image_shapes=data.get("original_image_shapes"), protocol=self.protocol, **kwargs
+            )
+        except Exception as err:
+            error_str = f"Error during NimClient inference [{self.model_interface.name()}, {self.protocol}]: {err}"
+            logger.error(error_str)
 
-        # Perform inference
-        if self.protocol == "grpc":
-            logger.debug("Performing gRPC inference...")
-            response = self._grpc_infer(formatted_input, model_name)
-            logger.debug("gRPC inference received response")
-        elif self.protocol == "http":
-            logger.debug("Performing HTTP inference...")
-            response = self._http_infer(formatted_input)
-            logger.debug("HTTP inference received response")
-        else:
-            raise ValueError("Invalid protocol specified. Must be 'grpc' or 'http'.")
+            raise RuntimeError(error_str)
 
-        # Parse and process output
-        parsed_output = self.model_interface.parse_output(
-            response, protocol=self.protocol, data=prepared_data, **kwargs
-        )
-        results = self.model_interface.process_inference_results(
-            parsed_output, original_image_shapes=data.get("original_image_shapes"), protocol=self.protocol, **kwargs
-        )
         return results
 
     def _grpc_infer(self, formatted_input: np.ndarray, model_name: str) -> np.ndarray:
@@ -275,7 +282,7 @@ class NimClient:
                 if status_code == 429 or status_code == 503 or (500 <= status_code < 600):
                     logger.warning(
                         f"Received HTTP {status_code} ({response.reason}) from "
-                        f"{self.model_interface.name()}. Attempt {attempt+1} of {self.max_retries}."
+                        f"{self.model_interface.name()}. Attempt {attempt + 1} of {self.max_retries}."
                     )
                     if attempt == self.max_retries - 1:
                         # No more retries left
@@ -297,12 +304,12 @@ class NimClient:
                 # Treat timeouts similarly to 5xx => attempt a retry
                 logger.warning(
                     f"HTTP request timed out after {self.timeout} seconds during {self.model_interface.name()} "
-                    f"inference. Attempt {attempt+1} of {self.max_retries}."
+                    f"inference. Attempt {attempt + 1} of {self.max_retries}."
                 )
                 if attempt == self.max_retries - 1:
                     logger.error("Max retries exceeded after repeated timeouts.")
                     raise TimeoutError(
-                        f"Repeated timeouts for {self.model_interface.name()} after {attempt+1} attempts."
+                        f"Repeated timeouts for {self.model_interface.name()} after {attempt + 1} attempts."
                     )
                 # Exponential backoff
                 backoff_time = base_delay * (2**attempt)
diff --git a/src/nv_ingest/util/nim/paddle.py b/src/nv_ingest/util/nim/paddle.py
index 5883f26..f844a94 100644
--- a/src/nv_ingest/util/nim/paddle.py
+++ b/src/nv_ingest/util/nim/paddle.py
@@ -22,19 +22,16 @@ class PaddleOCRModelInterface(ModelInterface):
     An interface for handling inference with a PaddleOCR model, supporting both gRPC and HTTP protocols.
     """
 
-    def __init__(
-        self,
-        paddle_version: Optional[str] = None,
-    ):
+    def __init__(self, paddle_version: Optional[str] = None) -> None:
         """
         Initialize the PaddleOCR model interface.
 
         Parameters
         ----------
         paddle_version : str, optional
-            The version of the PaddleOCR model (default: None).
+            The version of the PaddleOCR model (default is None).
         """
-        self.paddle_version = paddle_version
+        self.paddle_version: Optional[str] = paddle_version
 
     def name(self) -> str:
         """
@@ -49,48 +46,83 @@ class PaddleOCRModelInterface(ModelInterface):
 
     def prepare_data_for_inference(self, data: Dict[str, Any]) -> Dict[str, Any]:
         """
-        Prepare input data by decoding one or more base64-encoded images into numpy arrays.
+        Decode one or more base64-encoded images into NumPy arrays, storing them
+        alongside their dimensions in `data`.
 
         Parameters
         ----------
-        data : dict
+        data : dict of str -> Any
             The input data containing either:
-             - 'base64_image': a single base64-encoded image, OR
+             - 'base64_image': a single base64-encoded image, or
              - 'base64_images': a list of base64-encoded images.
 
         Returns
         -------
-        dict
-            The updated data dictionary with "image_arrays", a list of decoded image arrays.
+        dict of str -> Any
+            The updated data dictionary with the following keys added:
+            - "image_arrays": List of decoded NumPy arrays of shape (H, W, C).
+            - "image_dims": List of (height, width) tuples for each decoded image.
+
+        Raises
+        ------
+        KeyError
+            If neither 'base64_image' nor 'base64_images' is found in `data`.
+        ValueError
+            If 'base64_images' is present but is not a list.
         """
         if "base64_images" in data:
             base64_list = data["base64_images"]
             if not isinstance(base64_list, list):
                 raise ValueError("The 'base64_images' key must contain a list of base64-encoded strings.")
 
-            image_arrays = []
-            self._dims = []
+            image_arrays: List[np.ndarray] = []
+            dims: List[Tuple[int, int]] = []
             for b64 in base64_list:
                 img = base64_to_numpy(b64)
                 image_arrays.append(img)
-                self._dims.append((img.shape[0], img.shape[1]))  # (height, width)
+                dims.append((img.shape[0], img.shape[1]))
+
             data["image_arrays"] = image_arrays
+            data["image_dims"] = dims
 
         elif "base64_image" in data:
             # Single-image fallback
             img = base64_to_numpy(data["base64_image"])
             data["image_arrays"] = [img]
-            self._dims = [(img.shape[0], img.shape[1])]  # store one pair
+            data["image_dims"] = [(img.shape[0], img.shape[1])]
 
         else:
             raise KeyError("Input data must include 'base64_image' or 'base64_images'.")
 
         return data
 
-    def format_input(self, data: Dict[str, Any], protocol: str, **kwargs) -> Any:
+    def format_input(self, data: Dict[str, Any], protocol: str, **kwargs: Any) -> Any:
         """
-        Format input data for the specified protocol ("grpc" or "http"),
-        now capable of batching multiple images.
+        Format input data for the specified protocol ("grpc" or "http"), supporting batched data.
+
+        Parameters
+        ----------
+        data : dict of str -> Any
+            The input data dictionary, expected to contain "image_arrays" (list of np.ndarray).
+        protocol : str
+            The inference protocol, either "grpc" or "http".
+        **kwargs : Any
+            Additional keyword arguments for future use.
+
+        Returns
+        -------
+        Any
+            The formatted data ready to be sent via the specified protocol. For gRPC,
+            a batched NumPy array of shape (B, H, W, C). For HTTP, a JSON-serializable
+            payload containing the base64 images in the format required by the PaddleOCR endpoint.
+
+        Raises
+        ------
+        KeyError
+            If "image_arrays" is not found in `data`.
+        ValueError
+            If an invalid protocol is specified, or if the image shapes are inconsistent
+            for gRPC batching.
         """
         if "image_arrays" not in data:
             raise KeyError("Expected 'image_arrays' in data. Call prepare_data_for_inference first.")
@@ -100,21 +132,16 @@ class PaddleOCRModelInterface(ModelInterface):
         if protocol == "grpc":
             logger.debug("Formatting input for gRPC PaddleOCR model (batched).")
 
-            # For each image in the batch:
-            # 1) Preprocess (if needed)
-            # 2) Cast to float32
-            # 3) Expand dims so shape => (1, H, W, C)
-            processed = []
+            processed: List[np.ndarray] = []
             for img in images:
                 arr = preprocess_image_for_paddle(img, self.paddle_version).astype(np.float32)
                 arr = np.expand_dims(arr, axis=0)  # => shape (1, H, W, C)
                 processed.append(arr)
 
-            # Check that all images (beyond the batch dimension) have the same shape
-            # If not, raise an error
+            # Check that all images have the same shape (excluding batch dimension)
             shapes = [p.shape[1:] for p in processed]  # List of (H, W, C) shapes
             if not all(s == shapes[0] for s in shapes[1:]):
-                raise ValueError(f"All images must have the same dimensions for gRPC batching. " f"Found: {shapes}")
+                raise ValueError(f"All images must have the same dimensions for gRPC batching. Found: {shapes}")
 
             # Concatenate along the batch dimension => shape (B, H, W, C)
             batched_input = np.concatenate(processed, axis=0)
@@ -123,14 +150,12 @@ class PaddleOCRModelInterface(ModelInterface):
         elif protocol == "http":
             logger.debug("Formatting input for HTTP PaddleOCR model (batched).")
 
-            # For HTTP, we build a single payload that includes ALL images.
-            # Distinguish between legacy vs. new API:
             if self._is_version_early_access_legacy_api():
                 # Legacy => {"messages":[{"content":[imageObj, imageObj, ...]}]}
-                content_list = []
+                content_list: List[Dict[str, Any]] = []
                 base64_list = data.get("base64_images")
                 if base64_list is None and "base64_image" in data:
-                    # fallback to single
+                    # fallback to single image
                     base64_list = [data["base64_image"]]
 
                 for b64 in base64_list:
@@ -142,11 +167,11 @@ class PaddleOCRModelInterface(ModelInterface):
                 payload = {"messages": [message]}
 
             else:
-                # New => {"input":[ {"type":"image_url","url":...}, {"type":"image_url","url":...}, ... ]}
-                input_list = []
+                # New => {"input":[ {"type":"image_url","url":...}, ... ]}
+                input_list: List[Dict[str, Any]] = []
                 base64_list = data.get("base64_images")
                 if base64_list is None and "base64_image" in data:
-                    # fallback to single
+                    # fallback to single image
                     base64_list = [data["base64_image"]]
 
                 for b64 in base64_list:
@@ -157,12 +182,35 @@ class PaddleOCRModelInterface(ModelInterface):
                 payload = {"input": input_list}
 
             return payload
+
         else:
             raise ValueError("Invalid protocol specified. Must be 'grpc' or 'http'.")
 
-    def parse_output(self, response: Any, protocol: str, data: Optional[Dict[str, Any]] = None, **kwargs) -> Any:
+    def parse_output(self, response: Any, protocol: str, data: Optional[Dict[str, Any]] = None, **kwargs: Any) -> Any:
         """
-        Parse the output from the model's inference response. For multi-image gRPC or HTTP,
+        Parse the model's inference response for the given protocol. The parsing
+        may handle batched outputs for multiple images.
+
+        Parameters
+        ----------
+        response : Any
+            The raw response from the PaddleOCR model.
+        protocol : str
+            The protocol used for inference, "grpc" or "http".
+        data : dict of str -> Any, optional
+            Additional data dictionary that may include "image_dims" for bounding box scaling.
+        **kwargs : Any
+            Additional keyword arguments, such as custom `table_content_format`.
+
+        Returns
+        -------
+        Any
+            The parsed output, typically a list of (content, table_content_format) tuples.
+
+        Raises
+        ------
+        ValueError
+            If an invalid protocol is specified.
         """
         default_table_content_format = (
             TableFormatEnum.SIMPLE if self._is_version_early_access_legacy_api() else TableFormatEnum.PSEUDO_MARKDOWN
@@ -177,44 +225,64 @@ class PaddleOCRModelInterface(ModelInterface):
             )
             table_content_format = TableFormatEnum.SIMPLE
 
+        # Retrieve image dimensions if available
+        dims: Optional[List[Tuple[int, int]]] = data.get("image_dims") if data else None
+
         if protocol == "grpc":
             logger.debug("Parsing output from gRPC PaddleOCR model (batched).")
-            return self._extract_content_from_paddle_grpc_response(response, table_content_format)
+            return self._extract_content_from_paddle_grpc_response(response, table_content_format, dims)
 
         elif protocol == "http":
             logger.debug("Parsing output from HTTP PaddleOCR model (batched).")
-            return self._extract_content_from_paddle_http_response(response, table_content_format)
+            return self._extract_content_from_paddle_http_response(response, table_content_format, dims)
 
         else:
             raise ValueError("Invalid protocol specified. Must be 'grpc' or 'http'.")
 
-    def process_inference_results(self, output: Any, **kwargs) -> Any:
+    def process_inference_results(self, output: Any, **kwargs: Any) -> Any:
         """
         Process inference results for the PaddleOCR model.
 
         Parameters
         ----------
         output : Any
-            The raw output from the model.
-        kwargs : dict
-            Additional parameters for processing.
+            The raw output parsed from the PaddleOCR model.
+        **kwargs : Any
+            Additional keyword arguments for customization.
 
         Returns
         -------
         Any
-            The processed inference results.
+            The post-processed inference results. By default, this simply returns the output
+            as the table content (or content list).
         """
-
-        # For PaddleOCR, the output is the table content as a string
         return output
 
-    def _is_version_early_access_legacy_api(self):
-        return self.paddle_version and (pkgversion.parse(self.paddle_version) < pkgversion.parse("0.2.1-rc2"))
+    def _is_version_early_access_legacy_api(self) -> bool:
+        """
+        Determine if the current PaddleOCR version is considered "early access" and thus uses
+        the legacy API format.
+
+        Returns
+        -------
+        bool
+            True if the version is < 0.2.1-rc2; False otherwise.
+        """
+        return self.paddle_version is not None and pkgversion.parse(self.paddle_version) < pkgversion.parse("0.2.1-rc2")
 
     def _prepare_paddle_payload(self, base64_img: str) -> Dict[str, Any]:
         """
-        DEPRECATED by batch logic in format_input.
-        (Kept here if you need single-image direct calls.)
+        DEPRECATED by batch logic in format_input. Kept here if you need single-image direct calls.
+
+        Parameters
+        ----------
+        base64_img : str
+            A single base64-encoded image string.
+
+        Returns
+        -------
+        dict of str -> Any
+            The payload in either legacy or new format for PaddleOCR's HTTP endpoint.
         """
         image_url = f"data:image/png;base64,{base64_img}"
 
@@ -229,23 +297,45 @@ class PaddleOCRModelInterface(ModelInterface):
         return payload
 
     def _extract_content_from_paddle_http_response(
-        self, json_response: Dict[str, Any], table_content_format: Optional[str]
+        self, json_response: Dict[str, Any], table_content_format: Optional[str], dims: Optional[List[Tuple[int, int]]]
     ) -> List[Tuple[str, str]]:
         """
         Extract content from the JSON response of a PaddleOCR HTTP API request.
-        Always return a list of (content, table_content_format) tuples.
+
+        Parameters
+        ----------
+        json_response : dict of str -> Any
+            The JSON response returned by the PaddleOCR endpoint.
+        table_content_format : str or None
+            The specified format for table content (e.g., 'simple' or 'pseudo_markdown').
+        dims : list of (int, int), optional
+            A list of (height, width) for each corresponding image, used for bounding box
+            scaling if not None.
+
+        Returns
+        -------
+        list of (str, str)
+            A list of (content, table_content_format) tuples, one for each image result.
+
+        Raises
+        ------
+        RuntimeError
+            If the response format is missing or invalid.
+        ValueError
+            If the `table_content_format` is unrecognized.
         """
         if "data" not in json_response or not json_response["data"]:
             raise RuntimeError("Unexpected response format: 'data' key is missing or empty.")
 
-        results = []
+        results: List[str] = []
         for item_idx, item in enumerate(json_response["data"]):
             if self._is_version_early_access_legacy_api():
                 content = item.get("content", "")
             else:
                 text_detections = item.get("text_detections", [])
-                text_predictions = []
-                bounding_boxes = []
+                text_predictions: List[str] = []
+                bounding_boxes: List[List[Tuple[float, float]]] = []
+
                 for td in text_detections:
                     text_predictions.append(td["text_prediction"]["text"])
                     bounding_boxes.append([(pt["x"], pt["y"]) for pt in td["bounding_box"]["points"]])
@@ -254,69 +344,86 @@ class PaddleOCRModelInterface(ModelInterface):
                     content = " ".join(text_predictions)
                 elif table_content_format == TableFormatEnum.PSEUDO_MARKDOWN:
                     content = self._convert_paddle_response_to_psuedo_markdown(
-                        bounding_boxes, text_predictions, img_index=item_idx
+                        bounding_boxes, text_predictions, img_index=item_idx, dims=dims
                     )
                 else:
                     raise ValueError(f"Unexpected table format: {table_content_format}")
 
             results.append(content)
 
-        # Convert each content into a tuple (content, format). Always return a list of such tuples.
+        # Convert each content into a tuple (content, format).
         return [(content, table_content_format) for content in results]
 
     def _extract_content_from_paddle_grpc_response(
-        self, response: np.ndarray, table_content_format: str
+        self, response: np.ndarray, table_content_format: str, dims: Optional[List[Tuple[int, int]]]
     ) -> List[Tuple[str, str]]:
         """
-        Parses a gRPC response for one or more images.
-
-        The response can have two possible shapes:
+        Parse a gRPC response for one or more images. The response can have two possible shapes:
           - (3,) for batch_size=1
           - (3, n) for batch_size=n
 
         In either case:
           response[0, i]: byte string containing bounding box data
           response[1, i]: byte string containing text prediction data
-          response[2, i]: (Optional) additional data/metadata (ignored or logged here)
+          response[2, i]: (Optional) additional data/metadata (ignored here)
 
-        Returns a list of (content, table_content_format) of length n.
+        Parameters
+        ----------
+        response : np.ndarray
+            The raw NumPy array from gRPC. Expected shape: (3,) or (3, n).
+        table_content_format : str
+            The format of the output text content, e.g. 'simple' or 'pseudo_markdown'.
+        dims : list of (int, int), optional
+            A list of (height, width) for each corresponding image, used for bounding box scaling.
+
+        Returns
+        -------
+        list of (str, str)
+            A list of (content, table_content_format) for each image.
+
+        Raises
+        ------
+        ValueError
+            If the response is not a NumPy array or has an unexpected shape,
+            or if the `table_content_format` is unrecognized.
         """
         if not isinstance(response, np.ndarray):
             raise ValueError("Unexpected response format: response is not a NumPy array.")
 
-        # If we have shape (3,), convert to (3,1) so we can handle everything uniformly
+        # If we have shape (3,), convert to (3, 1)
         if response.ndim == 1 and response.shape == (3,):
             response = response.reshape(3, 1)
         elif response.ndim != 2 or response.shape[0] != 3:
-            raise ValueError(f"Unexpected response shape: {response.shape}. " "Expecting (3,) or (3, n).")
+            raise ValueError(f"Unexpected response shape: {response.shape}. Expecting (3,) or (3, n).")
 
         batch_size = response.shape[1]
-        results = []
+        results: List[Tuple[str, str]] = []
 
         for i in range(batch_size):
             # 1) Parse bounding boxes
-            bboxes_bytestr = response[0, i]
+            bboxes_bytestr: bytes = response[0, i]
             bounding_boxes = json.loads(bboxes_bytestr.decode("utf8"))
 
             # 2) Parse text predictions
-            texts_bytestr = response[1, i]
+            texts_bytestr: bytes = response[1, i]
             text_predictions = json.loads(texts_bytestr.decode("utf8"))
 
-            # 3) Optionally handle or log the third element (extra data/metadata)
-            extra_data_bytestr = response[2, i]
+            # 3) Log the third element (extra data/metadata) if needed
+            extra_data_bytestr: bytes = response[2, i]
             logger.debug(f"Ignoring extra_data for image {i}: {extra_data_bytestr}")
 
+            # Some gRPC responses nest single-item lists; flatten them if needed
             if isinstance(bounding_boxes, list) and len(bounding_boxes) == 1:
                 bounding_boxes = bounding_boxes[0]
             if isinstance(text_predictions, list) and len(text_predictions) == 1:
                 text_predictions = text_predictions[0]
 
-            # Construct the content string based on the desired format
+            # Construct the content string
             if table_content_format == TableFormatEnum.SIMPLE:
                 content = " ".join(text_predictions)
             elif table_content_format == TableFormatEnum.PSEUDO_MARKDOWN:
                 content = self._convert_paddle_response_to_psuedo_markdown(
-                    bounding_boxes, text_predictions, img_index=i
+                    bounding_boxes, text_predictions, img_index=i, dims=dims
                 )
             else:
                 raise ValueError(f"Unexpected table format: {table_content_format}")
@@ -325,51 +432,95 @@ class PaddleOCRModelInterface(ModelInterface):
 
         return results
 
+    @staticmethod
     def _convert_paddle_response_to_psuedo_markdown(
-        self, bounding_boxes: List[Any], text_predictions: List[str], img_index: int = 0
+        bounding_boxes: List[Any],
+        text_predictions: List[str],
+        img_index: int = 0,
+        dims: Optional[List[Tuple[int, int]]] = None,
     ) -> str:
         """
-        Convert bounding boxes & text to pseudo-markdown. For multiple images,
-        we use self._dims[img_index] to recover the correct height/width.
+        Convert bounding boxes & text to pseudo-markdown format. For multiple images,
+        the correct image dimensions (height, width) are retrieved from `dims[img_index]`.
+
+        Parameters
+        ----------
+        bounding_boxes : list of Any
+            A list (per line of text) of bounding boxes, each a list of (x, y) points.
+        text_predictions : list of str
+            A list of text predictions, one for each bounding box.
+        img_index : int, optional
+            The index of the image for which bounding boxes are being converted. Default is 0.
+        dims : list of (int, int), optional
+            A list of (height, width) for each corresponding image.
+
+        Returns
+        -------
+        str
+            The pseudo-markdown representation of detected text lines and bounding boxes.
+            Each cluster of text is placed on its own line, with text columns separated by '|'.
+
+        Notes
+        -----
+        - If `dims` is None or `img_index` is out of range, bounding boxes will not be scaled properly.
         """
-        if img_index >= len(self._dims):
-            logger.warning("Image index out of range for stored dimensions. Using first image dims by default.")
-            target_h, target_w = self._dims[0]
+        # Default to no scaling if dims are missing or out of range
+        if not dims:
+            logger.warning("No image_dims provided; bounding boxes will not be scaled.")
+            target_h, target_w = 1, 1
         else:
-            target_h, target_w = self._dims[img_index]
+            if img_index >= len(dims):
+                logger.warning("Image index out of range for stored dimensions. Using first image dims by default.")
+                target_h, target_w = dims[0]
+            else:
+                target_h, target_w = dims[img_index]
 
-        bboxes = []
-        texts = []
+        scaled_boxes: List[List[float]] = []
+        texts: List[str] = []
+
+        # Convert normalized coords back to actual pixel coords
         for box, txt in zip(bounding_boxes, text_predictions):
             if box == "nan":
                 continue
-            points = []
+            points: List[List[float]] = []
             for point in box:
-                # Convert normalized coords back to actual pixel coords
                 x = float(point[0]) * target_w
                 y = float(point[1]) * target_h
                 points.append([x, y])
-            bboxes.append(points)
+            scaled_boxes.append(points)
             texts.append(txt)
 
-        if not bboxes or not texts:
+        if not scaled_boxes or not texts:
             return ""
 
-        bboxes = np.array(bboxes).astype(int)
-        bboxes = bboxes.reshape(-1, 8)[:, [0, 1, 2, -1]]
+        # Convert bounding boxes to a simplified (x0, y0, x1, y1) representation
+        # by taking only the top-left and bottom-right corners
+        bboxes_array = np.array(scaled_boxes).astype(int)
+        # Reshape => (N, 4) by taking [0,1, 2,3] from the original (N, 4, 2)
+        # but we have 4 corners => 8 values. So shape => (N, 8). Then keep indices [0, 1, 2, 7].
+        bboxes_array = bboxes_array.reshape(-1, 8)[:, [0, 1, 2, -1]]
 
         preds_df = pd.DataFrame(
-            {"x0": bboxes[:, 0], "y0": bboxes[:, 1], "x1": bboxes[:, 2], "y1": bboxes[:, 3], "text": texts}
+            {
+                "x0": bboxes_array[:, 0],
+                "y0": bboxes_array[:, 1],
+                "x1": bboxes_array[:, 2],
+                "y1": bboxes_array[:, 3],
+                "text": texts,
+            }
         )
+        # Sort by top position
         preds_df = preds_df.sort_values("y0")
 
         dbscan = DBSCAN(eps=10, min_samples=1)
         dbscan.fit(preds_df["y0"].values[:, None])
 
         preds_df["cluster"] = dbscan.labels_
+        # Sort by cluster and then by x0 to group text lines from left to right
         preds_df = preds_df.sort_values(["cluster", "x0"])
 
-        results = ""
+        lines = []
         for _, dfg in preds_df.groupby("cluster"):
-            results += "| " + " | ".join(dfg["text"].values.tolist()) + " |\n"
-        return results
+            lines.append("| " + " | ".join(dfg["text"].values.tolist()) + " |")
+
+        return "\n".join(lines)


Test patch:
diff --git a/tests/nv_ingest/util/nim/test_helpers.py b/tests/nv_ingest/util/nim/test_helpers.py
index f0fae03..993e55d 100644
--- a/tests/nv_ingest/util/nim/test_helpers.py
+++ b/tests/nv_ingest/util/nim/test_helpers.py
@@ -213,7 +213,7 @@ def test_nimclient_infer_invalid_protocol(mock_model_interface, both_endpoints):
     client = NimClient(mock_model_interface, "grpc", both_endpoints)
     client.protocol = "invalid_protocol"
 
-    with pytest.raises(ValueError, match="Invalid protocol specified. Must be 'grpc' or 'http'."):
+    with pytest.raises(RuntimeError, match=".*Invalid protocol specified. Must be 'grpc' or 'http'\\..*"):
         client.infer({}, model_name="test_model")
 
 
diff --git a/tests/nv_ingest/util/nim/test_paddle.py b/tests/nv_ingest/util/nim/test_paddle.py
index ada46a2..136429c 100644
--- a/tests/nv_ingest/util/nim/test_paddle.py
+++ b/tests/nv_ingest/util/nim/test_paddle.py
@@ -113,8 +113,7 @@ def test_prepare_data_for_inference(paddle_ocr_model):
         assert len(result["image_arrays"]) == 1
         assert result["image_arrays"][0].shape == (100, 100, 3)
 
-        # We also store dimensions in self._dims
-        assert paddle_ocr_model._dims == [(100, 100)]
+        assert data["image_dims"][0] == (100, 100)
 
 
 def test_format_input_grpc(paddle_ocr_model):
@@ -190,7 +189,7 @@ def test_parse_output_http_pseudo_markdown(paddle_ocr_model, mock_paddle_http_re
     result = paddle_ocr_model.parse_output(mock_paddle_http_response, protocol="http")
     # It's a list with one tuple => (content, format).
     assert len(result) == 1
-    assert result[0][0] == "| mock_text |\n"
+    assert result[0][0] == "| mock_text |"
     assert result[0][1] == "pseudo_markdown"