diff --git a/src/dstack/_internal/cli/models/preset_agent.py b/src/dstack/_internal/cli/models/preset_agent.py index 6bd4e622b9..857e67206a 100644 --- a/src/dstack/_internal/cli/models/preset_agent.py +++ b/src/dstack/_internal/cli/models/preset_agent.py @@ -9,11 +9,16 @@ ValidationError, ValidatorFunctionWrapHandler, WrapValidator, + model_validator, ) from dstack._internal.core.models.common import CoreModel from dstack._internal.core.models.configurations import ServiceConfiguration -from dstack._internal.core.models.presets import PresetBenchmark +from dstack._internal.core.models.presets import ( + PresetBenchmark, + PresetBenchmarkLatency, + PresetBenchmarkMetrics, +) class PresetAgentInvalidService(CoreModel): @@ -42,6 +47,33 @@ def _service_from_yaml(value: Any, handler: ValidatorFunctionWrapHandler) -> Any ] +class PresetAgentBenchmarkMetrics(PresetBenchmarkMetrics): + """Fresh reports require the E2E latency missing from legacy stored presets.""" + + e2e_ms: PresetBenchmarkLatency + + +class PresetAgentBenchmark(PresetBenchmark): + metrics: PresetAgentBenchmarkMetrics + + @model_validator(mode="after") + def validate_reported_tpot_is_possible(self) -> "PresetAgentBenchmark": + tpot_ms = self.metrics.tpot_ms + if tpot_ms is None: + return self + decode_token_count = self.metrics.total_output_tokens - self.metrics.successful_requests + reported_decode_seconds = decode_token_count * tpot_ms.mean / 1000 + available_slot_seconds = self.metrics.duration_seconds * self.workload.concurrency + # Benchmark tools round their displayed latency values. A one-percent + # allowance prevents that formatting from rejecting a boundary value. + if reported_decode_seconds > available_slot_seconds * 1.01: + raise ValueError( + "reported tpot_ms exceeds the decode time available from duration_seconds" + " and workload.concurrency" + ) + return self + + class PresetAgentSuccess(CoreModel): """A preset the agent created, cross-checked against the run in `verify`.""" @@ -53,7 +85,7 @@ class PresetAgentSuccess(CoreModel): base: Annotated[str, Field(min_length=1)] model: Annotated[str, Field(min_length=1)] context_length: PositiveInt - benchmark: PresetBenchmark + benchmark: PresetAgentBenchmark class PresetAgentFailure(CoreModel): diff --git a/src/dstack/_internal/cli/services/presets/output.py b/src/dstack/_internal/cli/services/presets/output.py index 05db1828ce..b48bf8220e 100644 --- a/src/dstack/_internal/cli/services/presets/output.py +++ b/src/dstack/_internal/cli/services/presets/output.py @@ -211,9 +211,9 @@ def _add_session(table: Table, session: dict[str, Any], *, verbose: bool = False tps = _format_number(best["tok_s"]) if objective: # Lead with per-user tps: comparable across rows regardless of concurrency. - tpot_ms = best.get("tpot_ms") - if isinstance(tpot_ms, (int, float)) and tpot_ms > 0: - parts.append(f"tps/user={_format_number(1000 / tpot_ms)}") + per_user_tok_s = best.get("per_user_tok_s") + if isinstance(per_user_tok_s, (int, float)) and per_user_tok_s > 0: + parts.append(f"tps/user={_format_number(per_user_tok_s)}") if verbose: parts.append(f"tps={tps}") ttft_ms = best.get("ttft_ms") diff --git a/src/dstack/_internal/cli/services/presets/resources/system_prompt.md b/src/dstack/_internal/cli/services/presets/resources/system_prompt.md index 0ebc1e92b6..a108f0e135 100644 --- a/src/dstack/_internal/cli/services/presets/resources/system_prompt.md +++ b/src/dstack/_internal/cli/services/presets/resources/system_prompt.md @@ -367,6 +367,7 @@ trial benchmarks in `trials//trial.json`, the final benchmark as "total_input_tokens": 16384, "total_output_tokens": 2048, "output_tok_per_s": 512.0, "per_user_tok_per_s": 64.0, "ttft_ms": {"mean": 110.9, "p50": 108.2, "p99": 121.6}, + "e2e_ms": {"mean": 1063.4, "p50": 1048.0, "p99": 1150.3}, "tpot_ms": {"mean": 7.5, "p50": 7.4, "p99": 8.1} } } @@ -379,7 +380,13 @@ input and output token counts of the benchmark, rounded to whole tokens. Compute `output_tok_per_s` as `total_output_tokens / duration_seconds` and `per_user_tok_per_s` as `output_tok_per_s / workload.concurrency`. These are -the numbers used to compare trials (see `## Performance`). +the numbers used to compare trials (see `## Performance`). Record the benchmark +tool's per-request end-to-end latency as `e2e_ms`. Record `tpot_ms` only when the +tool reports time per output token; set it to `null` when the tool only reports +inter-token latency per streamed chunk. In particular, do not use SGLang's ITL +row as TPOT: speculative decoding can put several accepted tokens in one chunk. +The CLI derives its per-user decode rate from `e2e_ms`, `ttft_ms`, and the output +token totals rather than trusting either self-reported throughput field. Set `tool` to the command name and subcommands without options or values, `tool_version` to the exact version, and `command` to the secret-free diff --git a/src/dstack/_internal/cli/services/presets/session.py b/src/dstack/_internal/cli/services/presets/session.py index 68bd44446d..e1ce0823e7 100644 --- a/src/dstack/_internal/cli/services/presets/session.py +++ b/src/dstack/_internal/cli/services/presets/session.py @@ -628,12 +628,10 @@ def _trial_entry( gpu: Optional[str], ) -> dict[str, Any]: ttft = (metrics.get("ttft_ms") or {}).get("p50") - # Mean, not p50: MTP skews TPOT right, and mean is the delivered decode rate. - tpot = (metrics.get("tpot_ms") or {}).get("mean") context_length = record.get("context_length") return { "tok_s": tok_s, - "tpot_ms": tpot if isinstance(tpot, (int, float)) and tpot > 0 else None, + "per_user_tok_s": _effective_per_user_tok_s(metrics), "ttft_ms": ttft if isinstance(ttft, (int, float)) else None, "context_length": context_length if isinstance(context_length, int) else None, "concurrency": workload.get("concurrency"), @@ -641,6 +639,32 @@ def _trial_entry( } +def _effective_per_user_tok_s(metrics: dict[str, Any]) -> Optional[float]: + e2e_ms = (metrics.get("e2e_ms") or {}).get("mean") + ttft_ms = (metrics.get("ttft_ms") or {}).get("mean") + total_output_tokens = metrics.get("total_output_tokens") + successful_requests = metrics.get("successful_requests") + if ( + isinstance(e2e_ms, (int, float)) + and isinstance(ttft_ms, (int, float)) + and isinstance(total_output_tokens, (int, float)) + and isinstance(successful_requests, (int, float)) + and e2e_ms > ttft_ms + and successful_requests > 0 + and total_output_tokens > successful_requests + ): + mean_output_tokens = total_output_tokens / successful_requests + mean_tpot_ms = (e2e_ms - ttft_ms) / (mean_output_tokens - 1) + return 1000 / mean_tpot_ms + + # Sessions created by older clients have no E2E latency. Preserve their + # display using the reported TPOT, but new reports never depend on it. + legacy_tpot_ms = (metrics.get("tpot_ms") or {}).get("mean") + if isinstance(legacy_tpot_ms, (int, float)) and legacy_tpot_ms > 0: + return 1000 / legacy_tpot_ms + return None + + def _format_trial_gpu(record: dict[str, Any]) -> Optional[str]: resources = record.get("resources") gpu = resources.get("gpu") if isinstance(resources, dict) else None diff --git a/src/dstack/_internal/cli/services/presets/verify.py b/src/dstack/_internal/cli/services/presets/verify.py index 36fcda5fd7..d6d99fb223 100644 --- a/src/dstack/_internal/cli/services/presets/verify.py +++ b/src/dstack/_internal/cli/services/presets/verify.py @@ -192,7 +192,9 @@ def _normalized_benchmark( {"dataset": None} if configuration.dataset is None else {"shared_prefix_tokens": 0} ) ) - return benchmark.model_copy(update={"workload": workload}) + return PresetBenchmark.model_validate( + {**benchmark.model_dump(), "workload": workload.model_dump()} + ) def _portable_service( diff --git a/src/dstack/_internal/core/models/presets.py b/src/dstack/_internal/core/models/presets.py index 95e87fb29f..87df83066c 100644 --- a/src/dstack/_internal/core/models/presets.py +++ b/src/dstack/_internal/core/models/presets.py @@ -83,13 +83,22 @@ class PresetBenchmarkMetrics(CoreModel): output_tok_per_s: PositiveFloat per_user_tok_per_s: PositiveFloat ttft_ms: PresetBenchmarkLatency - tpot_ms: PresetBenchmarkLatency + # Old stored presets predate `e2e_ms`. New agent reports require it via + # `PresetAgentBenchmarkMetrics` and use it for the derived decode rate. + e2e_ms: Optional[PresetBenchmarkLatency] = None + # Some benchmark tools report inter-token latency per streamed chunk rather + # than TPOT per output token. Keep genuine TPOT when available, but do not + # force the agent to substitute a different metric. + tpot_ms: Optional[PresetBenchmarkLatency] = None class PresetBenchmark(CoreModel): - """The agent reports its benchmark in exactly this shape, and is forced to by - the schema generated from it. Changing a field here means also changing the - `## Benchmark` section of the system prompt, which tells it what to put there.""" + """The benchmark shape stored in presets. + + Fresh agent reports use a stricter subclass in `cli/models/preset_agent.py`. + Changing these fields also means changing the `## Benchmark` section of the + system prompt, which tells the agent what to report. + """ tool: Annotated[str, Field(min_length=1)] tool_version: Annotated[str, Field(min_length=1)] @@ -103,7 +112,15 @@ def effective_output_tok_per_s(self) -> float: @property def effective_per_user_tok_per_s(self) -> float: - return 1000 / self.metrics.tpot_ms.mean + e2e_ms = self.metrics.e2e_ms + if e2e_ms is None: + # Compatibility with presets stored before end-to-end latency was + # recorded. New agent reports always provide `e2e_ms`. + assert self.metrics.tpot_ms is not None + return 1000 / self.metrics.tpot_ms.mean + mean_output_tokens = self.metrics.total_output_tokens / self.metrics.successful_requests + mean_tpot_ms = (e2e_ms.mean - self.metrics.ttft_ms.mean) / (mean_output_tokens - 1) + return 1000 / mean_tpot_ms @field_validator("command") @classmethod @@ -125,6 +142,14 @@ def validate_metrics(self) -> Self: raise ValueError("benchmark must not include failed requests") if self.metrics.successful_requests != self.workload.num_requests: raise ValueError("benchmark request count must match workload.num_requests") + if self.metrics.e2e_ms is None: + if self.metrics.tpot_ms is None: + raise ValueError("legacy benchmark without e2e_ms must include tpot_ms") + return self + if self.metrics.total_output_tokens <= self.metrics.successful_requests: + raise ValueError("benchmark must average more than one output token per request") + if self.metrics.e2e_ms.mean <= self.metrics.ttft_ms.mean: + raise ValueError("benchmark mean e2e_ms must be greater than mean ttft_ms") return self diff --git a/src/tests/_internal/cli/common.py b/src/tests/_internal/cli/common.py index 35ae8bb785..47f24c807d 100644 --- a/src/tests/_internal/cli/common.py +++ b/src/tests/_internal/cli/common.py @@ -114,6 +114,7 @@ def get_preset_benchmark() -> PresetBenchmark: "output_tok_per_s": 42.1, "per_user_tok_per_s": 42.1, "ttft_ms": {"mean": 110.9, "p50": 108.2, "p99": 121.6}, + "e2e_ms": {"mean": 1063.4, "p50": 1048.0, "p99": 1150.3}, "tpot_ms": {"mean": 7.5, "p50": 7.4, "p99": 8.1}, }, ) @@ -254,5 +255,5 @@ def get_successful_preset_report(run: Run) -> PresetAgentSuccess: base="Qwen/Qwen3.5-27B", model="community/Qwen3.5-27B-GPTQ-Int4", context_length=32768, - benchmark=get_preset_benchmark(), + benchmark=get_preset_benchmark().model_dump(), ) diff --git a/src/tests/_internal/cli/services/presets/test_agent.py b/src/tests/_internal/cli/services/presets/test_agent.py index f101258c57..c86a25ec65 100644 --- a/src/tests/_internal/cli/services/presets/test_agent.py +++ b/src/tests/_internal/cli/services/presets/test_agent.py @@ -1096,13 +1096,33 @@ def test_counts_records_even_when_trials_share_a_task(self, tmp_path): assert summary["count"] == 4 assert summary["best"] == { "tok_s": 2300.0, - "tpot_ms": None, + "per_user_tok_s": None, "ttft_ms": None, "context_length": None, "concurrency": 8, "gpu": "A40:48GB:1", } + def test_derives_per_user_rate_from_e2e_instead_of_reported_tpot(self, tmp_path): + record = { + "benchmark": { + "metrics": { + "successful_requests": 4, + "duration_seconds": 10.0, + "total_output_tokens": 512, + "ttft_ms": {"mean": 100.0}, + "e2e_ms": {"mean": 1370.0}, + # Chunk-level ITL from speculative decoding, mislabeled as TPOT. + "tpot_ms": {"mean": 137.18}, + }, + "workload": {"concurrency": 4}, + } + } + + summary = _summarize_session_trials(_write_trials(tmp_path, [record])) + + assert summary["best"]["per_user_tok_s"] == pytest.approx(100.0) + def test_trials_are_ordered_numerically_beyond_nine(self, tmp_path): # Twelve trials: a string sort would chart 1, 10, 11, 12, 2, ... records = [ diff --git a/src/tests/_internal/cli/services/presets/test_create.py b/src/tests/_internal/cli/services/presets/test_create.py index e3ae148d88..77a1b2de75 100644 --- a/src/tests/_internal/cli/services/presets/test_create.py +++ b/src/tests/_internal/cli/services/presets/test_create.py @@ -52,6 +52,7 @@ from dstack._internal.core.errors import CLIError from dstack._internal.core.models.configurations import PresetConfiguration from dstack._internal.core.models.envs import EnvSentinel +from dstack._internal.core.models.presets import PresetBenchmarkLatency from dstack._internal.core.models.runs import Run, RunStatus from tests._internal.cli.common import ( get_preset, @@ -730,11 +731,22 @@ def test_throughput_is_derived_not_read(self): benchmark = preset.benchmark benchmark.metrics.output_tok_per_s = 999999.0 benchmark.metrics.per_user_tok_per_s = 999999.0 + benchmark.metrics.tpot_ms = PresetBenchmarkLatency( + mean=999999.0, p50=999999.0, p99=999999.0 + ) expected = benchmark.metrics.total_output_tokens / benchmark.metrics.duration_seconds assert benchmark.effective_output_tok_per_s == expected - # Per-user speed is the steady decode rate, not the aggregate over concurrency. - assert benchmark.effective_per_user_tok_per_s == 1000 / benchmark.metrics.tpot_ms.mean + # Per-user speed is the steady decode rate derived from the E2E and TTFT + # means, not a self-reported rate or chunk-level ITL mislabeled as TPOT. + assert benchmark.metrics.e2e_ms is not None + mean_output_tokens = ( + benchmark.metrics.total_output_tokens / benchmark.metrics.successful_requests + ) + mean_tpot_ms = (benchmark.metrics.e2e_ms.mean - benchmark.metrics.ttft_ms.mean) / ( + mean_output_tokens - 1 + ) + assert benchmark.effective_per_user_tok_per_s == 1000 / mean_tpot_ms class TestBuildConstraints: diff --git a/src/tests/_internal/cli/services/presets/test_output.py b/src/tests/_internal/cli/services/presets/test_output.py index ce70c66f41..6e7d442418 100644 --- a/src/tests/_internal/cli/services/presets/test_output.py +++ b/src/tests/_internal/cli/services/presets/test_output.py @@ -32,10 +32,17 @@ def test_formats_second_scale_ttft_without_scientific_notation(self): ttft.mean = 8148.3 ttft.p50 = 8151.4 ttft.p99 = 8334.2 + e2e = preset.benchmark.metrics.e2e_ms + tpot = preset.benchmark.metrics.tpot_ms + assert e2e is not None and tpot is not None + output_token_intervals = preset.benchmark.workload.output_tokens - 1 + e2e.mean = ttft.mean + output_token_intervals * tpot.mean + e2e.p50 = ttft.p50 + output_token_intervals * tpot.p50 + e2e.p99 = ttft.p99 + output_token_intervals * tpot.p99 output = output_module.format_preset_benchmark(preset, verbose=True) - # Per-user speed is 1/TPOT (p50 7.4ms), not the aggregate over concurrency. + # Per-user speed is the derived mean decode rate, not aggregate throughput. assert output.startswith("tps/user=133 ") assert output_module.format_preset_objective(preset).startswith("[secondary]io=1K/128 ") assert "ctx=32K" in output @@ -171,7 +178,7 @@ def test_shows_progress_after_status_and_best_benchmark(self): "count": 2, "best": { "tok_s": 2339.0, - "tpot_ms": 3.42, + "per_user_tok_s": 292.0, "concurrency": 8, "gpu": "A40:48GB:1", }, @@ -181,8 +188,8 @@ def test_shows_progress_after_status_and_best_benchmark(self): # 2 completed, so the 3rd is the one being trialed. assert row["STATUS"] == "[bold sea_green3]trialing[/] [secondary](3/3)[/]" - # Per-user speed is 1/TPOT, the same definition the preset row uses — not - # the aggregate over concurrency, which would read 292 here. + # Per-user speed comes from the trial's E2E/TTFT-derived decode rate, not + # aggregate throughput. assert row["BENCHMARK"].startswith("tps/user=292") assert row["RESOURCES"] == "A40:48GB:1" @@ -441,8 +448,8 @@ def test_the_fastest_failed_trial_is_kept_when_nothing_passed(self, tmp_path): assert summary["best"] is None assert summary["best_failed"]["tok_s"] == 300.0 - # Mean, not p50: MTP skews TPOT right, and mean is the delivered rate. - assert summary["best_failed"]["tpot_ms"] == 41.7 + # Older sessions without E2E preserve their reported mean TPOT display. + assert summary["best_failed"]["per_user_tok_s"] == pytest.approx(1000 / 41.7) def test_a_run_that_met_nothing_still_shows_what_it_measured(self): row = _session_row( @@ -457,7 +464,7 @@ def test_a_run_that_met_nothing_still_shows_what_it_measured(self): "best": None, "best_failed": { "tok_s": 100.0, - "tpot_ms": 34.4, + "per_user_tok_s": 1000 / 34.4, "ttft_ms": 4300.0, "context_length": 1048576, "concurrency": 4, diff --git a/src/tests/_internal/cli/services/presets/test_verify.py b/src/tests/_internal/cli/services/presets/test_verify.py index 26750c7b78..299e808ae5 100644 --- a/src/tests/_internal/cli/services/presets/test_verify.py +++ b/src/tests/_internal/cli/services/presets/test_verify.py @@ -349,6 +349,28 @@ def test_redacts_known_secret_in_benchmark_command_instead_of_failing(self, tmp_ assert report.benchmark.command.endswith("Bearer [redacted]'") assert "sk-live" not in report.benchmark.command + def test_requires_e2e_latency_in_a_fresh_report(self, tmp_path): + run = get_running_service_run() + data = get_successful_preset_report(run).model_dump() + data["run_id"] = str(data["run_id"]) + data["benchmark"]["metrics"].pop("e2e_ms") + + with pytest.raises(CLIError, match="e2e_ms"): + self._load(tmp_path, data, redacted_values=()) + + def test_rejects_a_reported_tpot_that_exceeds_available_slot_time(self, tmp_path): + run = get_running_service_run() + data = get_successful_preset_report(run).model_dump() + data["run_id"] = str(data["run_id"]) + data["benchmark"]["metrics"]["tpot_ms"] = { + "mean": 137.18, + "p50": 55.34, + "p99": 1160.43, + } + + with pytest.raises(CLIError, match="exceeds the decode time available"): + self._load(tmp_path, data, redacted_values=()) + def test_still_rejects_unknown_bearer_token(self, tmp_path): run = get_running_service_run() data = get_successful_preset_report(run).model_dump() diff --git a/src/tests/_internal/core/models/test_presets.py b/src/tests/_internal/core/models/test_presets.py index e12a881663..4c44920561 100644 --- a/src/tests/_internal/core/models/test_presets.py +++ b/src/tests/_internal/core/models/test_presets.py @@ -106,6 +106,22 @@ def test_applies_every_rule(self, paths: List[str], error: Optional[str]): validate_preset_file_path(paths[0]) +class TestPresetBenchmarkMetrics: + def test_legacy_benchmark_uses_stored_tpot(self): + benchmark = PresetBenchmark.model_validate(get_benchmark_data(get_workload_data())) + + assert benchmark.metrics.e2e_ms is None + assert benchmark.effective_per_user_tok_per_s == 1000 / 7.5 + + def test_rejects_nonpositive_decode_interval(self): + data = get_benchmark_data(get_workload_data()) + data["metrics"]["e2e_ms"] = {"mean": 100.0, "p50": 100.0, "p99": 100.0} + data["metrics"]["ttft_ms"] = {"mean": 100.0, "p50": 90.0, "p99": 110.0} + + with pytest.raises(ValidationError, match="e2e_ms must be greater"): + PresetBenchmark.model_validate(data) + + class TestPresetBenchmarkWorkload: """`api` is a Literal, not a plain string: the agent-facing JSON schema is generated from these models. `dataset` states the requested dataset, and its