From d897852258d9bcc05101c54dcd0973d111de9723 Mon Sep 17 00:00:00 2001 From: amy-why-3459 Date: Wed, 17 Jun 2026 10:31:48 +0800 Subject: [PATCH] [BugFix] Fix the issue of incorrect generated tokens (#4461) Signed-off-by: amy-why-3459 --- tests/benchmarks/metrics/test_metrics.py | 31 +++++ tests/benchmarks/patch/test_patch.py | 167 ++++++++++++++++++++++- vllm_omni/benchmarks/metrics/metrics.py | 17 ++- vllm_omni/benchmarks/patch/patch.py | 113 ++++++++++++--- 4 files changed, 304 insertions(+), 24 deletions(-) diff --git a/tests/benchmarks/metrics/test_metrics.py b/tests/benchmarks/metrics/test_metrics.py index 2688b224f..32d9d7958 100644 --- a/tests/benchmarks/metrics/test_metrics.py +++ b/tests/benchmarks/metrics/test_metrics.py @@ -14,6 +14,37 @@ from vllm_omni.benchmarks.patch.patch import MixRequestFuncOutput pytestmark = [pytest.mark.core_model, pytest.mark.benchmark, pytest.mark.cpu] +def test_tpot_matches_mean_itl_per_request(): + """TPOT and pooled ITL should agree when output_len tracks ITL samples.""" + output = MixRequestFuncOutput() + output.success = True + output.prompt_len = 100 + # Simulate server reporting more tokens than SSE chunks (bundled tokens). + # len(itl)=2 → 2 inter-chunk intervals, but server generated 10 tokens. + output.output_tokens = 10 + output.generated_text = "hello world" + output.ttft = 0.05 + output.text_latency = 0.25 + output.latency = 0.30 + output.itl = [0.10, 0.10] + + metrics, _ = calculate_metrics( + input_requests=[], + outputs=[output], + dur_s=1.0, + tokenizer=None, + selected_percentiles=[50.0], + goodput_config_dict={}, + task_type=TaskType.GENERATION, + selected_percentile_metrics=["tpot", "itl"], + max_concurrency=None, + request_rate=float("inf"), + benchmark_duration=1.0, + ) + + assert metrics.mean_tpot_ms == pytest.approx(metrics.mean_itl_ms, rel=1e-6, abs=1e-6) + + def _make_output(prompt_len: int, output_tokens: int = 10) -> MixRequestFuncOutput: """Build a minimal successful MixRequestFuncOutput for metrics aggregation.""" output = MixRequestFuncOutput() diff --git a/tests/benchmarks/patch/test_patch.py b/tests/benchmarks/patch/test_patch.py index 35a18aea3..ac3aef31e 100644 --- a/tests/benchmarks/patch/test_patch.py +++ b/tests/benchmarks/patch/test_patch.py @@ -133,7 +133,9 @@ async def test_output_tokens_not_assigned_without_metrics(mocker: MockerFixture) # Assert assert output.success is True - assert output.output_tokens == 0, "output_tokens should default to 0 when no metrics" + # Without server metrics we no longer infer count from SSE chunks because + # one chunk may carry multiple tokens. output_tokens stays 0 here. + assert output.output_tokens == 0, "output_tokens should be 0 when no server metrics are present" assert output.generated_text == "Hello world" @@ -277,7 +279,7 @@ async def test_output_tokens_with_missing_num_tokens_out(mocker: MockerFixture): # Assert assert output.success is True - assert output.output_tokens == 0, "output_tokens should default to 0 when num_tokens_out is missing" + assert output.output_tokens == 0, "output_tokens should be 0 when num_tokens_out is missing" @pytest.mark.asyncio @@ -579,6 +581,167 @@ class TestTextLatencyAttribute: # ============================================================================ +@pytest.mark.asyncio +async def test_skips_empty_role_chunk_for_ttft_and_itl(mocker: MockerFixture): + """Role/empty text chunks must not count toward TTFT or ITL.""" + request_input = RequestFuncInput( + model="test-model", + model_name="test-model", + prompt="test prompt", + api_url="http://test.com/v1/chat/completions", + prompt_len=10, + output_len=20, + ) + + chunks = [ + create_sse_chunk( + { + "choices": [{"delta": {"role": "assistant", "content": ""}}], + "modality": "text", + } + ), + create_sse_chunk( + { + "choices": [{"delta": {"content": "Hello"}}], + "modality": "text", + "usage": {"prompt_tokens": 10, "completion_tokens": 1, "total_tokens": 11}, + } + ), + create_sse_chunk( + { + "choices": [{"delta": {"content": " world"}}], + "modality": "text", + "usage": {"prompt_tokens": 10, "completion_tokens": 2, "total_tokens": 12}, + } + ), + b"data: [DONE]\n\n", + ] + + mock_response = MockResponse(200, chunks, delay_between_chunks=0.01) + mock_session = mocker.AsyncMock() + mock_session.post = mocker.MagicMock(return_value=mock_response) + + output = await async_request_openai_chat_omni_completions(request_input, mock_session) + + assert output.success is True + assert output.output_tokens == 2 + assert len(output.itl) == 1 + assert output.ttft > 0 + assert output.text_latency - output.ttft == pytest.approx(sum(output.itl), rel=1e-6, abs=1e-6) + + +@pytest.mark.asyncio +async def test_bundled_tokens_split_itl_from_usage(mocker: MockerFixture): + """Multiple tokens in one SSE chunk should expand ITL via usage deltas.""" + request_input = RequestFuncInput( + model="test-model", + model_name="test-model", + prompt="test prompt", + api_url="http://test.com/v1/chat/completions", + prompt_len=10, + output_len=20, + ) + + chunks = [ + create_sse_chunk( + { + "choices": [{"delta": {"content": "A"}}], + "modality": "text", + "usage": {"prompt_tokens": 10, "completion_tokens": 1, "total_tokens": 11}, + } + ), + create_sse_chunk( + { + "choices": [{"delta": {"content": "BCD"}}], + "modality": "text", + "usage": {"prompt_tokens": 10, "completion_tokens": 4, "total_tokens": 14}, + } + ), + b"data: [DONE]\n\n", + ] + + mock_response = MockResponse(200, chunks, delay_between_chunks=0.02) + mock_session = mocker.AsyncMock() + mock_session.post = mocker.MagicMock(return_value=mock_response) + + output = await async_request_openai_chat_omni_completions(request_input, mock_session) + + assert output.success is True + assert output.output_tokens == 4 + assert len(output.itl) == 3 + assert output.text_latency - output.ttft == pytest.approx(sum(output.itl), rel=1e-6, abs=1e-6) + + +@pytest.mark.asyncio +async def test_output_tokens_prefers_usage_over_metrics(mocker: MockerFixture): + request_input = RequestFuncInput( + model="test-model", + model_name="test-model", + prompt="test prompt", + api_url="http://test.com/v1/chat/completions", + prompt_len=10, + output_len=20, + ) + + chunks = [ + create_sse_chunk( + { + "choices": [{"delta": {"content": "Hello"}}], + "modality": "text", + "metrics": {"num_tokens_out": 5}, + "usage": {"prompt_tokens": 10, "completion_tokens": 8, "total_tokens": 18}, + } + ), + b"data: [DONE]\n\n", + ] + + mock_response = MockResponse(200, chunks) + mock_session = mocker.AsyncMock() + mock_session.post = mocker.MagicMock(return_value=mock_response) + + output = await async_request_openai_chat_omni_completions(request_input, mock_session) + + assert output.success is True + assert output.output_tokens == 8 + + +@pytest.mark.asyncio +async def test_metrics_only_chunk_updates_output_tokens(mocker: MockerFixture): + request_input = RequestFuncInput( + model="test-model", + model_name="test-model", + prompt="test prompt", + api_url="http://test.com/v1/chat/completions", + prompt_len=10, + output_len=20, + ) + + chunks = [ + create_sse_chunk( + { + "choices": [{"delta": {"content": "Text response"}}], + "modality": "text", + } + ), + create_sse_chunk( + { + "modality": "text", + "metrics": {"num_tokens_out": 25, "num_tokens_in": 10}, + } + ), + b"data: [DONE]\n\n", + ] + + mock_response = MockResponse(200, chunks) + mock_session = mocker.AsyncMock() + mock_session.post = mocker.MagicMock(return_value=mock_response) + + output = await async_request_openai_chat_omni_completions(request_input, mock_session) + + assert output.success is True + assert output.output_tokens == 25 + + @pytest.mark.asyncio async def test_prompt_len_assigned_from_usage(mocker: MockerFixture): # Arrange: request claims prompt_len=100, but server reports 4992 (multimodal). diff --git a/vllm_omni/benchmarks/metrics/metrics.py b/vllm_omni/benchmarks/metrics/metrics.py index 100ab19ef..b60bae347 100644 --- a/vllm_omni/benchmarks/metrics/metrics.py +++ b/vllm_omni/benchmarks/metrics/metrics.py @@ -768,11 +768,18 @@ def calculate_metrics( total_input += outputs[i].prompt_len tpot = 0 if output_len > 1: - try: - latency_minus_ttft = outputs[i].text_latency - outputs[i].ttft - except Exception: - latency_minus_ttft = outputs[i].latency - outputs[i].ttft - tpot = latency_minus_ttft / (output_len - 1) + if outputs[i].itl: + # Use mean(ITL) directly so per-request TPOT == mean(ITL). + # The ITL list records one entry per SSE chunk; server may + # bundle multiple tokens per chunk, so len(itl)+1 != output_len. + # Using mean(itl) keeps TPOT and ITL on the same footing. + tpot = sum(outputs[i].itl) / len(outputs[i].itl) + else: + try: + latency_minus_ttft = outputs[i].text_latency - outputs[i].ttft + except Exception: + latency_minus_ttft = outputs[i].latency - outputs[i].ttft + tpot = latency_minus_ttft / (output_len - 1) tpots.append(tpot) # Note: if output_len <= 1, we regard tpot as 0 for goodput all_tpots.append(tpot) diff --git a/vllm_omni/benchmarks/patch/patch.py b/vllm_omni/benchmarks/patch/patch.py index 73720f4bd..0ed06fa4b 100644 --- a/vllm_omni/benchmarks/patch/patch.py +++ b/vllm_omni/benchmarks/patch/patch.py @@ -530,6 +530,53 @@ def _extract_output_tokens_from_metrics(metrics: dict[str, Any]) -> int | None: return max(fallback_tokens, default=None) +def _apply_usage_to_output(output: MixRequestFuncOutput, usage: dict[str, Any]) -> int | None: + """Apply OpenAI ``usage`` fields to the benchmark output.""" + if (pt := _coerce_positive_int(usage.get("prompt_tokens"))) is not None: + output.prompt_len = pt + completion_tokens = _coerce_positive_int(usage.get("completion_tokens")) + if completion_tokens is not None: + output.output_tokens = max(int(output.output_tokens or 0), completion_tokens) + return completion_tokens + + +def _resolve_token_delta_from_usage( + completion_tokens: int | None, + completion_tokens_seen: int, +) -> tuple[int, int]: + if completion_tokens is None or completion_tokens <= completion_tokens_seen: + return 0, completion_tokens_seen + delta = completion_tokens - completion_tokens_seen + return delta, completion_tokens + + +def _record_text_token_stream_intervals( + output: MixRequestFuncOutput, + *, + timestamp: float, + start_time: float, + token_delta: int, + most_recent_timestamp: float, +) -> float: + """Record TTFT/ITL for ``token_delta`` newly generated text tokens.""" + if token_delta <= 0: + return most_recent_timestamp + + if output.ttft == 0.0: + output.ttft = timestamp - start_time + output.text_latency = timestamp - start_time + most_recent_timestamp = timestamp + if token_delta > 1: + output.itl.extend([0.0] * (token_delta - 1)) + return most_recent_timestamp + + interval = max(timestamp - most_recent_timestamp, 0.0) + per_token = interval / token_delta + output.itl.extend([per_token] * token_delta) + output.text_latency = timestamp - start_time + return timestamp + + def _update_output_stage_metrics_from_payload( output: MixRequestFuncOutput, data: dict[str, Any], @@ -541,7 +588,7 @@ def _update_output_stage_metrics_from_payload( return if update_output_tokens: if (num_tokens_out := _extract_output_tokens_from_metrics(metrics)) is not None: - output.output_tokens = num_tokens_out + output.output_tokens = max(int(output.output_tokens or 0), num_tokens_out) if isinstance(sid := metrics.get("stage_id"), int): output.stage_id = sid if isinstance(final_output_type := metrics.get("final_output_type"), str): @@ -624,6 +671,14 @@ async def async_request_openai_chat_omni_completions( "stream": True, "stream_options": { "include_usage": True, + # Per-chunk completion_tokens lets _resolve_token_delta_from_usage + # compute the exact token count for each SSE flush. Without this, + # one SSE chunk can carry multiple tokens (asyncio coalescing), so + # len(itl)+1 < actual_tokens and ITL measures per-chunk latency + # (~1.74× tokens_per_chunk) rather than true per-token latency. + # NOTE: the vLLM StreamOptions field is "continuous_usage_stats", + # NOT "include_continuous_usage". + "continuous_usage_stats": True, }, } _update_payload_common(payload, request_func_input) @@ -661,7 +716,6 @@ async def async_request_openai_chat_omni_completions( first_inconsistent_wav_params: tuple[int, int, int] | None = None # For non-wav responses, accumulate encoded bytes then decode once. audio_bytes_buffer = bytearray() - ttft = 0.0 st = time.perf_counter() output.start_time = st most_recent_timestamp = st @@ -685,6 +739,7 @@ async def async_request_openai_chat_omni_completions( output.image_generation_time_ms = 0.0 output.image_pixels = 0 output.denoise_step_latency_ms = 0.0 + completion_tokens_seen = 0 try: async with session.post(url=api_url, json=payload, headers=headers) as response: if response.status == 200: @@ -712,22 +767,45 @@ async def async_request_openai_chat_omni_completions( if chunk != "[DONE]": timestamp = time.perf_counter() data = json.loads(chunk) + _update_output_stage_metrics_from_payload(output, data) + usage = data.get("usage") + completion_tokens = None + if isinstance(usage, dict): + completion_tokens = _apply_usage_to_output(output, usage) + if choices := data.get("choices"): modality = data.get("modality") - delta = choices[0].get("delta") or {} + choice = choices[0] + delta = choice.get("delta") or {} content = delta.get("content") if not content and isinstance(delta.get("audio"), dict): content = delta["audio"].get("data") if modality == "text": - # First token - if ttft == 0.0: - ttft = timestamp - st - output.ttft = ttft - else: - output.itl.append(timestamp - most_recent_timestamp) - generated_text += content or "" - most_recent_timestamp = timestamp - output.text_latency = timestamp - st + token_delta, completion_tokens_seen = _resolve_token_delta_from_usage( + completion_tokens, + completion_tokens_seen, + ) + token_ids = choice.get("token_ids") + if token_delta == 0 and token_ids: + token_delta = len(token_ids) + if completion_tokens is not None: + completion_tokens_seen = max( + completion_tokens_seen, + completion_tokens, + ) + has_text_content = bool(content) + if token_delta == 0 and has_text_content and completion_tokens is None: + token_delta = 1 + if token_delta > 0: + most_recent_timestamp = _record_text_token_stream_intervals( + output, + timestamp=timestamp, + start_time=st, + token_delta=token_delta, + most_recent_timestamp=most_recent_timestamp, + ) + if has_text_content: + generated_text += content elif modality == "audio": if output.audio_ttfp == 0.0: output.audio_ttfp = timestamp - st @@ -762,7 +840,6 @@ async def async_request_openai_chat_omni_completions( if content_image_ms > 0: output.image_generation_time_ms += content_image_ms - _update_output_stage_metrics_from_payload(output, data) ( metrics_image_count, metrics_image_ms, @@ -778,10 +855,6 @@ async def async_request_openai_chat_omni_completions( if metrics_denoise_step_ms > output.denoise_step_latency_ms: output.denoise_step_latency_ms = metrics_denoise_step_ms - if usage := data.get("usage"): - if (pt := usage.get("prompt_tokens")) is not None: - output.prompt_len = pt - if wav_inconsistent_chunk_count > 0: logger.warning( "Dropped %d wav chunks with inconsistent params during benchmark " @@ -794,6 +867,12 @@ async def async_request_openai_chat_omni_completions( output.latency = timestamp - st output.generated_text = generated_text + if output.itl: + # Align text_latency with ITL so TPOT formula and + # mean(ITL) are consistent. Do NOT infer output_tokens + # from len(itl)+1: one SSE chunk may carry multiple + # tokens, so the ITL count understates the real count. + output.text_latency = output.ttft + sum(output.itl) audio_duration_sec = 0.0 audio_frames = 0 if response_format == "wav" and wav_pcm_buffer and wav_audio_params is not None: