anodyine/vllm-omni
1
0 Fork 0
Code Issues 4 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

[BugFix] Fix the issue of incorrect generated tokens (#4461)

Browse Source 
Signed-off-by: amy-why-3459 <wuhaiyan17@huawei.com>
This commit is contained in:
amy-why-3459 authored and GitHub committed 2026-06-17 02:31:48 +00:00
1 parent b46c5af06f
commit d897852258
4 files changed
+304 -24
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/benchmarks/metrics/test_metrics.py
+31
View File
@@ -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()
tests/benchmarks/patch/test_patch.py
+165 -2
View File
@@ -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).
vllm_omni/benchmarks/metrics/metrics.py
+12 -5
View File
@@ -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)
vllm_omni/benchmarks/patch/patch.py
+96 -17
View File
@@ -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: