[NMCUR-421] Add inference-server PDF benchmark path

benchmarking/4xGB200-64CPU.yaml

@@ -234,6 +234,12 @@ entries:
   - name: nemotron_parse_pdf_raydata
     timeout_s: 2400
 
+  - name: nemotron_parse_pdf_inference_server_ray_serve
+    timeout_s: 2400
+
+  - name: nemotron_parse_pdf_inference_server_dynamo
+    timeout_s: 2400
+
   - name: alm_pipeline_xenna
     timeout_s: 1200
 

benchmarking/benchmarks.yaml

@@ -1666,17 +1666,6 @@ entries:
       --max-pages=10
       --max-num-seqs=64
     timeout_s: 1200
-    sink_data:
-      - name: slack
-        additional_metrics:
-          - num_pdfs_processed
-          - num_pages_processed
-          - num_output_tokens
-          - num_output_tasks
-          - throughput_pages_per_sec
-          - throughput_output_tokens_per_sec
-        ping_on_failure:
-          - U082G3E46R0  # Abhinav Garg
     requirements:
       # Exact counts: the dataset is fixed and every file in it renders, so a
       # correct run reproduces these precisely. Verified identical across three
@@ -1728,17 +1717,6 @@ entries:
       --max-pages=10
       --max-num-seqs=64
     timeout_s: 1200
-    sink_data:
-      - name: slack
-        additional_metrics:
-          - num_pdfs_processed
-          - num_pages_processed
-          - num_output_tokens
-          - num_output_tasks
-          - throughput_pages_per_sec
-          - throughput_output_tokens_per_sec
-        ping_on_failure:
-          - U082G3E46R0  # Abhinav Garg
     requirements:
       # Exact counts: the dataset is fixed and every file in it renders, so a
       # correct run reproduces these precisely. Verified identical across three
@@ -1773,6 +1751,71 @@ entries:
         min_value: 4630000
         max_value: 5115000
 
+  - name: nemotron_parse_pdf_inference_server_ray_serve
+    enabled: true
+    script: nemotron_parse_pdf_benchmark.py
+    args: >-
+      --benchmark-results-path={session_entry_dir}
+      --executor=ray_data
+      --manifest={dataset:nemotron_parse_pdf,manifest}
+      --pdf-dir={dataset:nemotron_parse_pdf,pdf_dir}
+      --output-dir={session_entry_dir}/scratch/output
+      --model-path={dataset:nemotron_parse_pdf_model,files}
+      --model-id=nvidia/NVIDIA-Nemotron-Parse-v1.2
+      --backend=vllm
+      --inference-server-type=ray-serve
+      --inference-server-client-workers-per-replica=4
+      --inference-batch-size=32
+      --enforce-eager
+      --pdfs-per-task=20
+      --max-pdfs=1278
+      --max-pages=10
+    timeout_s: 1800
+    requirements:
+      - metric: num_pdfs_processed
+        exact_value: 1278
+      - metric: num_pages_processed
+        exact_value: 5794
+      - metric: throughput_pages_per_sec
+        min_value: 7.0
+      - metric: throughput_output_tokens_per_sec
+        min_value: 6000
+      - metric: num_output_tokens
+        min_value: 4630000
+        max_value: 5115000
+
+  - name: nemotron_parse_pdf_inference_server_dynamo
+    enabled: true
+    script: nemotron_parse_pdf_benchmark.py
+    args: >-
+      --benchmark-results-path={session_entry_dir}
+      --executor=ray_data
+      --manifest={dataset:nemotron_parse_pdf,manifest}
+      --pdf-dir={dataset:nemotron_parse_pdf,pdf_dir}
+      --output-dir={session_entry_dir}/scratch/output
+      --model-path={dataset:nemotron_parse_pdf_model,files}
+      --model-id=nvidia/NVIDIA-Nemotron-Parse-v1.2
+      --backend=vllm
+      --inference-server-type=dynamo
+      --inference-server-client-workers-per-replica=4
+      --inference-batch-size=32
+      --enforce-eager
+      --pdfs-per-task=20
+      --max-pdfs=1278
+      --max-pages=10
+    timeout_s: 1800
+    requirements:
+      - metric: num_pdfs_processed
+        exact_value: 1278
+      - metric: num_pages_processed
+        exact_value: 5794
+      - metric: throughput_pages_per_sec
+        min_value: 7.0
+      - metric: throughput_output_tokens_per_sec
+        min_value: 6000
+      - metric: num_output_tokens
+        min_value: 4630000
+        max_value: 5115000
   # CPU-only; the 8-GPU runtime target does not apply.
   - name: alm_pipeline_xenna
     enabled: true

benchmarking/scripts/inference_server_utils.py

@@ -0,0 +1,136 @@
+# Copyright (c) 2026, NVIDIA CORPORATION.  All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# ruff: noqa: PLR0913
+
+"""Shared helpers for inference servers used by benchmark scripts."""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING, Any, Literal
+
+if TYPE_CHECKING:
+    from nemo_curator.core.serve import InferenceServer
+
+InferenceServerBackend = Literal["ray-serve", "dynamo"]
+
+
+def parse_json_object(value: str | None, *, argument: str) -> dict[str, Any]:
+    """Parse an optional command-line JSON object."""
+    if value is None:
+        return {}
+    try:
+        parsed = json.loads(value)
+    except json.JSONDecodeError as error:
+        msg = f"{argument} must be valid JSON: {error}"
+        raise ValueError(msg) from error
+    if not isinstance(parsed, dict):
+        msg = f"{argument} must decode to a JSON object"
+        raise TypeError(msg)
+    return parsed
+
+
+def static_num_replicas(autoscaling_config: dict[str, Any] | None) -> int:
+    """Resolve a fixed replica count from a Ray-style autoscaling config."""
+    if not autoscaling_config:
+        return 1
+    min_replicas = int(autoscaling_config.get("min_replicas", 1))
+    max_replicas = int(autoscaling_config.get("max_replicas", min_replicas))
+    if min_replicas != max_replicas:
+        msg = (
+            "Dynamo does not support autoscaling in benchmarks; "
+            f"min_replicas ({min_replicas}) must equal max_replicas ({max_replicas})."
+        )
+        raise ValueError(msg)
+    if min_replicas < 1:
+        msg = f"num_replicas must be at least 1, got {min_replicas}"
+        raise ValueError(msg)
+    return min_replicas
+
+
+def start_inference_server(
+    *,
+    backend: InferenceServerBackend,
+    model_id: str,
+    num_replicas: int,
+    engine_kwargs: dict[str, Any] | None = None,
+    model_path: str | None = None,
+    model_runtime_env: dict[str, Any] | None = None,
+    dynamo_kwargs: dict[str, Any] | None = None,
+    dynamo_router_kwargs: dict[str, Any] | None = None,
+    dynamo_subprocess_env: dict[str, str] | None = None,
+    ray_serve_deployment_config: dict[str, Any] | None = None,
+    health_check_timeout_s: int = 900,
+) -> InferenceServer:
+    """Build, start, and return an inference server.
+
+    If ``model_path`` is set, the server loads weights from that local path
+    while exposing ``model_id`` as the served model name.
+
+    ``model_runtime_env`` is passed to Ray Serve replicas or Dynamo workers.
+    For gpt-oss, set ``TIKTOKEN_RS_CACHE_DIR`` there to read the Harmony
+    encoding from a pre-populated local cache instead of downloading it from
+    Azure at startup (see https://github.com/openai/harmony/issues/101).
+
+    ``health_check_timeout_s`` controls how long server startup waits for the
+    model to register at ``/v1/models``.
+    """
+    from nemo_curator.core.serve import InferenceServer
+
+    if num_replicas < 1:
+        msg = f"num_replicas must be at least 1, got {num_replicas}"
+        raise ValueError(msg)
+    if backend == "dynamo":
+        from nemo_curator.core.serve import DynamoRouterConfig, DynamoServerConfig, DynamoVLLMModelConfig
+
+        model = DynamoVLLMModelConfig(
+            model_identifier=model_path or model_id,
+            model_name=model_id if model_path else None,
+            engine_kwargs=engine_kwargs or {},
+            num_replicas=num_replicas,
+            dynamo_kwargs=dynamo_kwargs or {},
+            runtime_env=model_runtime_env or {},
+        )
+        server = InferenceServer(
+            models=[model],
+            backend=DynamoServerConfig(
+                request_plane="tcp",
+                router=DynamoRouterConfig(router_kwargs=dynamo_router_kwargs or {}),
+                subprocess_env=dynamo_subprocess_env or {},
+            ),
+            health_check_timeout_s=health_check_timeout_s,
+        )
+    else:
+        if backend != "ray-serve":
+            msg = f"Unsupported inference server backend: {backend}"
+            raise ValueError(msg)
+
+        from nemo_curator.core.serve import RayServeModelConfig
+
+        model = RayServeModelConfig(
+            model_identifier=model_path or model_id,
+            model_name=model_id if model_path else None,
+            deployment_config=(
+                ray_serve_deployment_config
+                if ray_serve_deployment_config is not None
+                else {"num_replicas": num_replicas}
+            ),
+            engine_kwargs=engine_kwargs or {},
+            runtime_env=model_runtime_env or {},
+        )
+        server = InferenceServer(models=[model], health_check_timeout_s=health_check_timeout_s)
+
+    server.start()
+    return server