[Metrics][Spec Decoding] Add per-request acceptance rate Prometheus histogram

Track speculative decoding acceptance at the finished-request level.
Per-request draft and accepted token counts are carried through
EngineCoreOutput into IterationStats and FinishedRequestStats.

Expose vllm:request_spec_decode_acceptance_rate as a Prometheus
histogram observing each finished request's accepted/draft ratio.
The existing aggregate speculative decoding counters are preserved.

The scheduler now builds aggregate and request-level spec decode
stats together so invalid draft token handling stays consistent.

Add focused metrics stats tests for request-level histogram
observation and per-request spec decode stat accumulation.

Signed-off-by: jeff.ye <jeff.ye@novita.ai>

Author: jeffye-dev

tests/v1/metrics/test_stats.py

@@ -1,12 +1,33 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
-from vllm.v1.engine import FinishReason
+from unittest.mock import MagicMock
+
+from vllm.config import SpeculativeConfig
+from vllm.v1.engine import EngineCoreOutput, FinishReason
 from vllm.v1.metrics.stats import (
     IterationStats,
     PrefillStats,
     PromptTokenStats,
     RequestStateStats,
 )
+from vllm.v1.spec_decode.metrics import (
+    SpecDecodeRequestStats,
+    SpecDecodingProm,
+)
+
+
+class MockCounter:
+    def __init__(self, *args, **kwargs):
+        self.children = {}
+
+    def labels(self, *labelvalues):
+        child = MagicMock()
+        self.children[labelvalues] = child
+        return child
+
+
+class MockHistogram(MockCounter):
+    pass
 
 
 def test_iteration_stats_repr():
@@ -122,6 +143,98 @@ def test_prefill_kv_computed_edge_cases():
     assert finished_req2.request_id == "test-req-004"
 
 
+def test_finished_request_includes_spec_decode_stats():
+    iteration_stats = IterationStats()
+    req_stats = RequestStateStats(arrival_time=0.0)
+    req_stats.scheduled_ts = 0.1
+    req_stats.first_token_ts = 0.5
+    req_stats.last_token_ts = 1.0
+
+    for output in (
+        EngineCoreOutput(
+            request_id="test-req-005",
+            new_token_ids=[1, 2],
+            spec_decode_stats=SpecDecodeRequestStats(
+                num_draft_tokens=4,
+                num_accepted_tokens=1,
+            ),
+        ),
+        EngineCoreOutput(
+            request_id="test-req-005",
+            new_token_ids=[3, 4, 5],
+            spec_decode_stats=SpecDecodeRequestStats(
+                num_draft_tokens=4,
+                num_accepted_tokens=2,
+            ),
+        ),
+    ):
+        iteration_stats.update_from_output(
+            output=output,
+            engine_core_timestamp=2.0,
+            is_prefilling=False,
+            req_stats=req_stats,
+            lora_states=MagicMock(),
+            lora_name=None,
+        )
+
+    iteration_stats.update_from_finished_request(
+        finish_reason=FinishReason.STOP,
+        request_id="test-req-005",
+        num_prompt_tokens=2000,
+        max_tokens_param=300,
+        req_stats=req_stats,
+    )
+
+    finished_req = iteration_stats.finished_requests[0]
+    assert finished_req.spec_decode_stats is not None
+    assert finished_req.spec_decode_stats.num_draft_tokens == 8
+    assert finished_req.spec_decode_stats.num_accepted_tokens == 3
+    assert finished_req.spec_decode_stats.acceptance_rate == 3 / 8
+
+
+def test_spec_decode_request_stats_merge():
+    stats = SpecDecodeRequestStats(
+        num_draft_tokens=4,
+        num_accepted_tokens=1,
+    ).merge(
+        SpecDecodeRequestStats(
+            num_draft_tokens=4,
+            num_accepted_tokens=2,
+        )
+    )
+
+    assert stats.num_draft_tokens == 8
+    assert stats.num_accepted_tokens == 3
+    assert stats.acceptance_rate == 3 / 8
+    assert stats.merge(None) is stats
+
+
+def test_request_spec_decode_acceptance_rate_observed():
+    class MockSpecDecodingProm(SpecDecodingProm):
+        _counter_cls = MockCounter
+        _histogram_cls = MockHistogram
+
+    prom = MockSpecDecodingProm(
+        speculative_config=SpeculativeConfig(
+            method="ngram",
+            num_speculative_tokens=2,
+        ),
+        labelnames=["model_name", "engine"],
+        per_engine_labelvalues={0: ["test-model", "0"]},
+    )
+
+    prom.observe_finished_request(
+        spec_decode_stats=SpecDecodeRequestStats(
+            num_draft_tokens=200,
+            num_accepted_tokens=10,
+        ),
+    )
+
+    prom.histogram_request_spec_decode_acceptance_rate[
+        0
+    ].observe.assert_called_once_with(0.05)
+
+
 def test_prompt_token_stats_all_computed():
     """Test all tokens computed locally, no caching."""
     stats = PromptTokenStats()

vllm/v1/core/sched/scheduler.py

@@ -58,7 +58,7 @@
 from vllm.v1.outputs import DraftTokenIds, KVConnectorOutput, ModelRunnerOutput
 from vllm.v1.request import Request, RequestStatus, StreamingUpdate
 from vllm.v1.spec_decode.dynamic.utils import build_dynamic_sd_schedule_lookup
-from vllm.v1.spec_decode.metrics import SpecDecodingStats
+from vllm.v1.spec_decode.metrics import SpecDecodeRequestStats, SpecDecodingStats
 from vllm.v1.structured_output import StructuredOutputManager
 from vllm.v1.utils import record_function_or_nullcontext
 
@@ -1547,13 +1547,23 @@ def update_from_output(
             scheduled_spec_token_ids = (
                 scheduler_output.scheduled_spec_decode_tokens.get(req_id)
             )
+            spec_decode_request_stats: SpecDecodeRequestStats | None = None
             if scheduled_spec_token_ids and (
                 generated_token_ids or self.num_sampled_tokens_per_step == 0
             ):
                 num_draft_tokens = len(scheduled_spec_token_ids)
                 num_sampled = self.num_sampled_tokens_per_step
                 num_accepted = max(len(generated_token_ids) - num_sampled, 0)
                 num_rejected = num_draft_tokens - num_accepted
+                spec_decoding_stats, spec_decode_request_stats = (
+                    self.make_spec_decoding_stats(
+                        spec_decoding_stats,
+                        num_draft_tokens=num_draft_tokens,
+                        num_accepted_tokens=num_accepted,
+                        num_invalid_spec_tokens=scheduler_output.num_invalid_spec_tokens,
+                        request_id=req_id,
+                    )
+                )
                 # num_computed_tokens represents the number of tokens
                 # processed in the current step, considering scheduled
                 # tokens and rejections. If some tokens are rejected,
@@ -1565,14 +1575,6 @@ def update_from_output(
                 # the scheduled spec tokens count and so is similarly adjusted.
                 if request.num_output_placeholders > 0:
                     request.num_output_placeholders -= num_rejected
-                spec_decoding_stats = self.make_spec_decoding_stats(
-                    spec_decoding_stats,
-                    num_draft_tokens=num_draft_tokens,
-                    num_accepted_tokens=num_accepted,
-                    num_invalid_spec_tokens=scheduler_output.num_invalid_spec_tokens,
-                    request_id=req_id,
-                )
-
             # Free encoder inputs only after the step has actually executed.
             if request.has_encoder_inputs:
                 self._free_encoder_inputs(request)
@@ -1701,6 +1703,7 @@ def update_from_output(
                         trace_headers=request.trace_headers,
                         routed_experts=routed_experts,
                         num_nans_in_logits=request.num_nans_in_logits,
+                        spec_decode_stats=spec_decode_request_stats,
                     )
                 )
             else:
@@ -2265,17 +2268,24 @@ def make_spec_decoding_stats(
         num_accepted_tokens: int,
         num_invalid_spec_tokens: dict[str, int] | None,
         request_id: str,
-    ) -> SpecDecodingStats | None:
+    ) -> tuple[SpecDecodingStats | None, SpecDecodeRequestStats | None]:
         if not self.log_stats or not num_draft_tokens:
-            return None
+            return None, None
         if spec_decoding_stats is None:
             spec_decoding_stats = SpecDecodingStats.new(self.num_spec_tokens)
         if num_invalid_spec_tokens:
             num_draft_tokens -= num_invalid_spec_tokens.get(request_id, 0)
         spec_decoding_stats.observe_draft(
-            num_draft_tokens=num_draft_tokens, num_accepted_tokens=num_accepted_tokens
+            num_draft_tokens=num_draft_tokens,
+            num_accepted_tokens=num_accepted_tokens,
+        )
+        return (
+            spec_decoding_stats,
+            SpecDecodeRequestStats(
+                num_draft_tokens=num_draft_tokens,
+                num_accepted_tokens=num_accepted_tokens,
+            ),
         )
-        return spec_decoding_stats
 
     def shutdown(self) -> None:
         logger.debug_once("[shutdown] Scheduler: start")

vllm/v1/engine/__init__.py

@@ -18,6 +18,7 @@
 from vllm.v1.metrics.stats import PrefillStats, SchedulerStats
 from vllm.v1.outputs import LogprobsLists, LogprobsTensors
 from vllm.v1.serial_utils import UtilityResult
+from vllm.v1.spec_decode.metrics import SpecDecodeRequestStats
 
 # Type for pause_generation mode parameter.
 # - "abort": Abort all in-flight requests immediately (default).
@@ -197,6 +198,7 @@ class EngineCoreOutput(
     # The number of NaNs in logits.
     # A value greater than 0 indicates that the output is corrupted.
     num_nans_in_logits: int = 0
+    spec_decode_stats: SpecDecodeRequestStats | None = None
 
     @property
     def finished(self) -> bool:

vllm/v1/metrics/loggers.py

@@ -1221,6 +1221,11 @@ def record(
                     finished_request.max_tokens_param
                 )
 
+            self.spec_decoding_prom.observe_finished_request(
+                spec_decode_stats=finished_request.spec_decode_stats,
+                engine_idx=engine_idx,
+            )
+
     def record_sleep_state(self, sleep: int = 0, level: int = 0):
         awake = 1
         discard_all = 0

vllm/v1/metrics/stats.py

@@ -9,7 +9,7 @@
 import vllm.envs as envs
 from vllm.compilation.cuda_graph import CUDAGraphStat
 from vllm.v1.metrics.perf import PerfStats
-from vllm.v1.spec_decode.metrics import SpecDecodingStats
+from vllm.v1.spec_decode.metrics import SpecDecodeRequestStats, SpecDecodingStats
 
 if TYPE_CHECKING:
     from vllm.v1.engine import EngineCoreEvent, EngineCoreOutput, FinishReason
@@ -218,6 +218,7 @@ class RequestStateStats:
 
     # Track if this request is corrupted (NaNs in logits)
     is_corrupted: bool = False
+    spec_decode_stats: SpecDecodeRequestStats | None = None
 
 
 @dataclass
@@ -237,6 +238,7 @@ class FinishedRequestStats:
     mean_time_per_output_token: float = 0.0
     is_corrupted: bool = False
     num_cached_tokens: int = 0
+    spec_decode_stats: SpecDecodeRequestStats | None = None
 
 
 @dataclass
@@ -371,6 +373,10 @@ def update_from_output(
             req_stats.first_token_latency = first_token_latency
 
         req_stats.num_generation_tokens += num_new_generation_tokens
+        if output.spec_decode_stats is not None:
+            req_stats.spec_decode_stats = output.spec_decode_stats.merge(
+                req_stats.spec_decode_stats
+            )
 
         # Track if this request is corrupted (only check once per request)
         # Early exit if already marked as corrupted to avoid redundant checks
@@ -472,6 +478,7 @@ def update_from_finished_request(
             mean_time_per_output_token=mean_time_per_output_token,
             is_corrupted=req_stats.is_corrupted,
             num_cached_tokens=num_cached_tokens,
+            spec_decode_stats=req_stats.spec_decode_stats,
         )
         self.finished_requests.append(finished_req)
 

vllm/v1/spec_decode/metrics.py

@@ -3,6 +3,7 @@
 
 import time
 from dataclasses import dataclass, field
+from typing import NamedTuple
 
 import numpy as np
 import prometheus_client
@@ -14,6 +15,27 @@
 logger = init_logger(__name__)
 
 
+class SpecDecodeRequestStats(NamedTuple):
+    """Request-level speculative decoding stats."""
+
+    num_accepted_tokens: int
+    num_draft_tokens: int
+
+    @property
+    def acceptance_rate(self) -> float:
+        if self.num_draft_tokens == 0:
+            return float("nan")
+        return self.num_accepted_tokens / self.num_draft_tokens
+
+    def merge(self, other: "SpecDecodeRequestStats | None") -> "SpecDecodeRequestStats":
+        if other is None:
+            return self
+        return SpecDecodeRequestStats(
+            num_accepted_tokens=(self.num_accepted_tokens + other.num_accepted_tokens),
+            num_draft_tokens=self.num_draft_tokens + other.num_draft_tokens,
+        )
+
+
 @dataclass
 class SpecDecodingStats:
     """Per-step iteration decoding stats from scheduler.
@@ -196,6 +218,7 @@ class SpecDecodingProm:
     """
 
     _counter_cls = prometheus_client.Counter
+    _histogram_cls = prometheus_client.Histogram
 
     def __init__(
         self,
@@ -212,6 +235,34 @@ def __init__(
         if not self.spec_decoding_enabled:
             return
 
+        if not is_diffusion:
+            histogram_request_acceptance_rate = self._histogram_cls(
+                name="vllm:request_spec_decode_acceptance_rate",
+                documentation=(
+                    "Histogram of request-level speculative decoding draft "
+                    "acceptance rates for finished requests."
+                ),
+                buckets=[
+                    0.0,
+                    0.1,
+                    0.2,
+                    0.3,
+                    0.4,
+                    0.5,
+                    0.6,
+                    0.7,
+                    0.8,
+                    0.9,
+                    1.0,
+                ],
+                labelnames=labelnames,
+            )
+            self.histogram_request_spec_decode_acceptance_rate = (
+                create_metric_per_engine(
+                    histogram_request_acceptance_rate, per_engine_labelvalues
+                )
+            )
+
         if is_diffusion:
             counter_specs = [
                 ("vllm:diffusion_num_denoising_steps", "Number of denoising steps."),
@@ -279,3 +330,20 @@ def observe(self, spec_decoding_stats: SpecDecodingStats, engine_idx: int = 0):
             self.counter_spec_decode_num_accepted_tokens_per_pos.get(engine_idx, [])
         ):
             counter.inc(spec_decoding_stats.num_accepted_tokens_per_pos[pos])
+
+    def observe_finished_request(
+        self,
+        spec_decode_stats: SpecDecodeRequestStats | None,
+        engine_idx: int = 0,
+    ) -> None:
+        if (
+            not self.spec_decoding_enabled
+            or self.is_diffusion
+            or spec_decode_stats is None
+            or spec_decode_stats.num_draft_tokens <= 0
+        ):
+            return
+
+        self.histogram_request_spec_decode_acceptance_rate[engine_idx].observe(
+            spec_decode_stats.acceptance_rate
+        )