[LLM] add metric observations for vLLM and SGLang

doc/workloads/sglang.rst

@@ -128,6 +128,18 @@ The reported metric (``default``) is throughput. Additional supported metrics ar
 CloudAI also provides the scenario-level ``sglang_comparison`` report. It compares SGLang test runs in the scenario and
 uses ``bench_cmd_args`` values as comparison labels.
 
+Structured metrics
+~~~~~~~~~~~~~~~~~~
+
+.. csv-table::
+   :header: "Metric", "Unit", "Preferred", "Dimensions"
+
+   "``request_throughput``", "requests/s", "Higher", "None"
+   "``output_token_throughput``", "tokens/s", "Higher", "None"
+   "``ttft`` (time to first token)", "ms", "Lower", "``statistic``: mean, median, p99"
+   "``tpot`` (time per output token)", "ms", "Lower", "``statistic``: mean, median, p99"
+   "``accuracy``", "ratio (0–1)", "Higher", "None"
+
 
 Readiness health checks
 -----------------------

doc/workloads/vllm.rst

@@ -123,6 +123,18 @@ The reported metric (``default``) is throughput. Additional supported metrics ar
 CloudAI also provides the scenario-level ``vllm_comparison`` report. It compares vLLM test runs in the scenario and
 uses ``bench_cmd_args`` values as comparison labels.
 
+Structured metrics
+~~~~~~~~~~~~~~~~~~
+
+.. csv-table::
+   :header: "Metric", "Unit", "Preferred", "Dimensions"
+
+   "``request_throughput``", "requests/s", "Higher", "None"
+   "``output_token_throughput``", "tokens/s", "Higher", "None"
+   "``ttft`` (time to first token)", "ms", "Lower", "``statistic``: mean, median, p99"
+   "``tpot`` (time per output token)", "ms", "Lower", "``statistic``: mean, median, p99"
+   "``accuracy``", "ratio (0–1)", "Higher", "None"
+
 
 Controlling the Number of GPUs
 -------------------------------

src/cloudai/metrics.py

@@ -105,6 +105,7 @@ def validate_dimensions(cls, values: Mapping[str, Any]) -> MetricDimensions:
 BACKEND = DimensionDefinition("backend", "Backend", Annotated[str, Field(strict=True, min_length=1)])
 SOURCE_MEMORY = DimensionDefinition("source_memory", "Source memory", Annotated[str, Field(strict=True, min_length=1)])
 TARGET_MEMORY = DimensionDefinition("target_memory", "Target memory", Annotated[str, Field(strict=True, min_length=1)])
+STATISTIC = DimensionDefinition("statistic", "Statistic", Literal["mean", "median", "p99"])
 
 BANDWIDTH = MetricDefinition(
     key="bandwidth",
@@ -119,8 +120,38 @@ def validate_dimensions(cls, values: Mapping[str, Any]) -> MetricDimensions:
     direction=OptimizationDirection.MINIMIZE,
 )
 
-MetricCatalog._metrics = {metric.key: metric for metric in (BANDWIDTH, LATENCY)}
-MetricCatalog.register_metrics(BANDWIDTH, LATENCY)
+REQUEST_THROUGHPUT = MetricDefinition(
+    key="request_throughput",
+    display_name="Request throughput",
+    unit="requests/s",
+    direction=OptimizationDirection.MAXIMIZE,
+)
+OUTPUT_TOKEN_THROUGHPUT = MetricDefinition(
+    key="output_token_throughput",
+    display_name="Output token throughput",
+    unit="tokens/s",
+    direction=OptimizationDirection.MAXIMIZE,
+)
+TTFT = MetricDefinition(
+    key="ttft",
+    display_name="Time to first token",
+    unit="ms",
+    direction=OptimizationDirection.MINIMIZE,
+)
+TPOT = MetricDefinition(
+    key="tpot",
+    display_name="Time per output token",
+    unit="ms",
+    direction=OptimizationDirection.MINIMIZE,
+)
+ACCURACY = MetricDefinition(
+    key="accuracy",
+    display_name="Accuracy",
+    unit="ratio",
+    direction=OptimizationDirection.MAXIMIZE,
+)
+
+MetricCatalog.register_metrics(BANDWIDTH, LATENCY, REQUEST_THROUGHPUT, OUTPUT_TOKEN_THROUGHPUT, TTFT, TPOT, ACCURACY)
 MetricCatalog.register_dimensions(
     SIZE_BYTES,
     BATCH_SIZE,
@@ -130,6 +161,7 @@ def validate_dimensions(cls, values: Mapping[str, Any]) -> MetricDimensions:
     BACKEND,
     SOURCE_MEMORY,
     TARGET_MEMORY,
+    STATISTIC,
 )
 
 

src/cloudai/workloads/common/llm_serving.py

@@ -16,6 +16,7 @@
 
 from __future__ import annotations
 
+import math
 import re
 import shlex
 from abc import ABC, abstractmethod
@@ -27,6 +28,7 @@
 from rich.table import Table
 from typing_extensions import Self
 
+import cloudai.metrics
 from cloudai.core import METRIC_ERROR, DockerImage, HFModel, Installable, MetricValue, ReportGenerationStrategy
 from cloudai.models.workload import CmdArgs, TestDefinition
 from cloudai.systems.slurm import SlurmCommandGenStrategy
@@ -241,6 +243,32 @@ def tps_per_user(self) -> float | None:
         return self.throughput / self.concurrency
 
 
+def llm_serving_metric_observations(
+    results: LLMServingBenchReport | None,
+    accuracy: float | None = None,
+) -> list[cloudai.metrics.MetricObservation]:
+    """Use statistics to distinguish latency results; scalar results have no dimensions."""
+    observations: list[cloudai.metrics.MetricObservation] = []
+    if accuracy is not None and math.isfinite(accuracy):
+        observations.append(cloudai.metrics.MetricObservation(cloudai.metrics.ACCURACY, accuracy, {}))
+    if results is None or results.completed <= 0:
+        return observations
+
+    for metric, field in (
+        (cloudai.metrics.REQUEST_THROUGHPUT, "request_throughput"),
+        (cloudai.metrics.OUTPUT_TOKEN_THROUGHPUT, "output_throughput"),
+    ):
+        value = getattr(results, field)
+        if math.isfinite(value):
+            observations.append(cloudai.metrics.MetricObservation(metric, value, {}))
+    for metric in (cloudai.metrics.TTFT, cloudai.metrics.TPOT):
+        for statistic in ("mean", "median", "p99"):
+            value = getattr(results, f"{statistic}_{metric.key}_ms")
+            if math.isfinite(value):
+                observations.append(cloudai.metrics.MetricObservation(metric, value, {"statistic": statistic}))
+    return observations
+
+
 class LLMServingReportGenerationStrategy(ReportGenerationStrategy, Generic[TestDefT, ReportT], ABC):
     """Shared report generation strategy for LLM serving workloads."""
 

src/cloudai/workloads/sglang/sglang.py

@@ -23,14 +23,16 @@
 
 from pydantic import ConfigDict, Field, field_validator, model_validator
 
-from cloudai.core import JobStatusResult, TestRun
+import cloudai.metrics
+from cloudai.core import JobStatusResult, System, TestRun
 from cloudai.models.workload import CmdArgs
 from cloudai.workloads.common.llm_serving import (
     CustomBash,
     LLMServingArgs,
     LLMServingBenchReport,
     LLMServingCmdArgs,
     LLMServingTestDefinition,
+    llm_serving_metric_observations,
     validate_custom_bash_patterns,
 )
 
@@ -127,11 +129,22 @@ def was_run_successful(self, tr: TestRun) -> JobStatusResult:
             error_message=f"SGLang bench jsonl does not contain successful requests in {tr.output_path}.",
         )
 
+    def metric_observations(self, system: System, tr: TestRun) -> list[cloudai.metrics.MetricObservation]:
+        del system
+        results = parse_sglang_bench_output(tr.output_path / SGLANG_BENCH_JSONL_FILE)
+        accuracy = (
+            parse_sglang_semantic_accuracy(tr.output_path / SGLANG_SEMANTIC_EVAL_LOG_FILE)
+            if self.semantic_eval_cmd_args is not None
+            else None
+        )
+        return llm_serving_metric_observations(results, accuracy)
+
 
 class SGLangBenchReport(LLMServingBenchReport):
     """Parsed benchmark data from SGLang bench_serving output."""
 
     request_throughput: float
+    output_throughput: float
 
     @property
     def throughput(self) -> float:

src/cloudai/workloads/vllm/report_generation_strategy.py

@@ -28,6 +28,7 @@ class VLLMBenchReport(LLMServingBenchReport):
     """Report for vLLM benchmark results."""
 
     output_throughput: float
+    request_throughput: float
 
     @property
     def throughput(self) -> float:

src/cloudai/workloads/vllm/vllm.py

@@ -25,6 +25,7 @@
 
 from pydantic import ConfigDict, Field, field_validator
 
+import cloudai.metrics
 from cloudai.core import GitRepo, Installable, JobStatusResult, System, TestRun
 from cloudai.models.workload import CmdArgs
 from cloudai.workloads.common.llm_serving import (
@@ -35,6 +36,7 @@
     all_gpu_ids,
     calculate_decode_gpu_ids,
     calculate_prefill_gpu_ids,
+    llm_serving_metric_observations,
     validate_custom_bash_patterns,
 )
 
@@ -264,6 +266,14 @@ def was_run_successful(self, tr: TestRun) -> JobStatusResult:
             is_successful=False, error_message=f"vLLM bench log does not contain benchmark result in {tr.output_path}."
         )
 
+    def metric_observations(self, system: System, tr: TestRun) -> list[cloudai.metrics.MetricObservation]:
+        del system
+        from .report_generation_strategy import parse_vllm_bench_output
+
+        results = parse_vllm_bench_output(tr.output_path / VLLM_BENCH_JSON_FILE)
+        accuracy = parse_vllm_semantic_accuracy(tr.output_path) if self.semantic_eval_cmd_args is not None else None
+        return llm_serving_metric_observations(results, accuracy)
+
 
 @cache
 def parse_vllm_semantic_accuracy(output_path: Path) -> float | None:

tests/workloads/common/test_llm_serving.py

@@ -14,6 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import json
 from pathlib import Path
 from types import SimpleNamespace
 from typing import Any, cast
@@ -22,6 +23,7 @@
 from pydantic import Field
 from rich.table import Table
 
+import cloudai.metrics
 from cloudai.core import GitRepo, TestRun
 from cloudai.systems.slurm import SlurmSystem
 from cloudai.workloads.common.llm_serving import (
@@ -34,6 +36,8 @@
     all_gpu_ids,
     parse_gpu_ids,
 )
+from cloudai.workloads.sglang import SglangCmdArgs, SglangSemanticEvalCmdArgs, SglangTestDefinition
+from cloudai.workloads.vllm import VllmCmdArgs, VllmSemanticEvalCmdArgs, VllmTestDefinition
 
 
 class FakeLLMArgs(LLMServingArgs):
@@ -461,3 +465,122 @@ def test_used_gpus_count_counts_all_cluster_gpus_for_two_node_disaggregated_run(
     )
 
     assert strategy.used_gpus_count() == 8
+
+
+@pytest.fixture(params=["vllm", "sglang"])
+def serving_observation_run(request: pytest.FixtureRequest, tmp_path: Path) -> tuple[TestRun, Path, dict]:
+    backend = request.param
+    if backend == "vllm":
+        tdef = VllmTestDefinition(
+            name=backend,
+            description="Serving observations",
+            test_template_name=backend,
+            cmd_args=VllmCmdArgs(docker_image_url="image:latest", model="test/model"),
+        )
+    else:
+        tdef = SglangTestDefinition(
+            name=backend,
+            description="Serving observations",
+            test_template_name=backend,
+            cmd_args=SglangCmdArgs(docker_image_url="image:latest", model="test/model"),
+        )
+    tr = TestRun(name=backend, test=tdef, num_nodes=1, nodes=[], output_path=tmp_path)
+    result_path = tmp_path / ("vllm-bench.json" if backend == "vllm" else "sglang-bench.jsonl")
+    data = {
+        "num_prompts": 30,
+        "completed": 30,
+        "max_concurrency": 16,
+        "request_throughput": 10.0,
+        "output_throughput": 2400.0,
+        "mean_ttft_ms": 120.0,
+        "median_ttft_ms": 100.0,
+        "p99_ttft_ms": 200.0,
+        "mean_tpot_ms": 12.0,
+        "median_tpot_ms": 10.0,
+        "p99_tpot_ms": 20.0,
+    }
+    result_path.write_text(json.dumps(data) + "\n")
+    return tr, result_path, data
+
+
+def test_serving_metric_observations(slurm_system: SlurmSystem, serving_observation_run) -> None:
+    tr, _, _ = serving_observation_run
+    observations = tr.test.metric_observations(slurm_system, tr)
+
+    assert len(observations) == len({(o.metric.key, tuple(o.dimensions.items())) for o in observations}) == 8
+    assert {(o.metric.key, o.dimensions.get("statistic")): (o.value, o.metric.unit) for o in observations} == {
+        ("request_throughput", None): (10.0, "requests/s"),
+        ("output_token_throughput", None): (2400.0, "tokens/s"),
+        ("ttft", "mean"): (120.0, "ms"),
+        ("ttft", "median"): (100.0, "ms"),
+        ("ttft", "p99"): (200.0, "ms"),
+        ("tpot", "mean"): (12.0, "ms"),
+        ("tpot", "median"): (10.0, "ms"),
+        ("tpot", "p99"): (20.0, "ms"),
+    }
+    for observation in observations:
+        assert set(observation.dimensions) == ({"statistic"} if observation.metric.key in {"ttft", "tpot"} else set())
+
+    tr.test.cmd_args.model = "other/model"
+    tr.test.bench_cmd_args.max_concurrency = 32
+    assert tr.test.metric_observations(slurm_system, tr) == observations
+
+
+@pytest.mark.parametrize("result", ["missing", "invalid", "incomplete", "unsuccessful"])
+def test_serving_metric_observations_without_results(
+    slurm_system: SlurmSystem, serving_observation_run, result: str
+) -> None:
+    tr, result_path, data = serving_observation_run
+    if result == "missing":
+        result_path.unlink()
+    elif result == "invalid":
+        result_path.write_text("{invalid\n")
+    elif result == "incomplete":
+        result_path.write_text("{}\n")
+    else:
+        result_path.write_text(json.dumps({**data, "completed": 0}))
+
+    assert tr.test.metric_observations(slurm_system, tr) == []
+
+
+@pytest.mark.parametrize("field", ["request_throughput", "output_throughput"])
+def test_serving_metric_observations_requires_both_throughputs(
+    slurm_system: SlurmSystem, serving_observation_run, field: str
+) -> None:
+    tr, result_path, data = serving_observation_run
+    del data[field]
+    result_path.write_text(json.dumps(data))
+
+    assert tr.test.metric_observations(slurm_system, tr) == []
+
+
+def test_serving_metric_observations_nonfinite_values(slurm_system: SlurmSystem, serving_observation_run) -> None:
+    tr, result_path, data = serving_observation_run
+    data.update(mean_ttft_ms=float("nan"), output_throughput=float("inf"), max_concurrency=0)
+    result_path.write_text(json.dumps(data))
+
+    observations = tr.test.metric_observations(slurm_system, tr)
+
+    assert len(observations) == 6
+    assert all("max_concurrency" not in o.dimensions for o in observations)
+    assert all(o.metric.key != "output_token_throughput" for o in observations)
+    assert all((o.metric.key, o.dimensions.get("statistic")) != ("ttft", "mean") for o in observations)
+
+
+@pytest.mark.parametrize("evaluation", ["disabled", "missing", "valid"])
+def test_serving_metric_observations_accuracy(
+    slurm_system: SlurmSystem, serving_observation_run, evaluation: str
+) -> None:
+    tr, _, _ = serving_observation_run
+    if evaluation != "disabled":
+        tr.test.semantic_eval_cmd_args = VllmSemanticEvalCmdArgs() if tr.name == "vllm" else SglangSemanticEvalCmdArgs()
+    if evaluation != "missing":
+        (tr.output_path / f"{tr.name}-semantic-eval.log").write_text("Accuracy: 0.875\n")
+
+    observations = tr.test.metric_observations(slurm_system, tr)
+    accuracy = [o for o in observations if o.metric.key == "accuracy"]
+
+    assert len(observations) == len({(o.metric.key, tuple(o.dimensions.items())) for o in observations})
+    assert accuracy == (
+        [cloudai.metrics.MetricObservation(cloudai.metrics.ACCURACY, 0.875, {})] if evaluation == "valid" else []
+    )