From 6583006b84f63678c97bb615f52f67bf15789a3b Mon Sep 17 00:00:00 2001 From: githubgxll <1094462054@qq.com> Date: Sun, 11 Oct 2026 09:53:52 +0800 Subject: [PATCH 1/3] fix(sglang): emit cumulative streaming usage with cache counts per engine update - OutputOptions: add continuous_usage_stats flag surviving Rust serialization - decode handler: report cumulative prompt/completion/cache usage per engine meta_info update in aggregated and PD decode paths; retain known sparse cache values; stop consuming after cancellation - custom frontend: forward include_usage+continuous_usage_stats, emit usage-only chunks while parsers buffer text/tool syntax, attach usage to terminal tool payload frames - tests: frontend metrics + decode handler behavior regressions - docs: describe option semantics, version-consistency and billing rules --- dingo/frontend/sglang_processor.py | 41 ++- .../test_sglang_processor_metrics_unit.py | 218 +++++++++++- .../request_handlers/llm/decode_handler.py | 103 ++++-- .../tests/test_sglang_decode_handler.py | 309 ++++++++++++++++++ docs/backends/sglang/README.md | 43 +++ lib/llm/src/protocols/common.rs | 5 + lib/llm/src/protocols/openai.rs | 1 + 7 files changed, 668 insertions(+), 52 deletions(-) diff --git a/dingo/frontend/sglang_processor.py b/dingo/frontend/sglang_processor.py index 00033b1d15ac..faf3e1811253 100644 --- a/dingo/frontend/sglang_processor.py +++ b/dingo/frontend/sglang_processor.py @@ -66,7 +66,7 @@ def _cached_tokens_details_from_engine_data( """Extract and normalize cached_tokens_details from engine_data. The decode handler copies sglang's ``meta_info.cached_tokens_details`` into - ``engine_data.cached_tokens_details`` on the finish chunk. This normalizes + ``engine_data.cached_tokens_details`` on usage-bearing chunks. This normalizes the shape so the carrier payload matches sglang's native ``sglext`` field. Returns ``None`` when the detail is absent or malformed. """ @@ -97,6 +97,15 @@ def _cached_tokens_from_usage(usage: dict[str, Any] | None) -> int | None: return cached_tokens if isinstance(cached_tokens, int) else None +def _continuous_usage_requested(request: dict[str, Any]) -> bool: + options = request.get("stream_options") or {} + return bool( + request.get("stream") + and options.get("include_usage") + and options.get("continuous_usage_stats") + ) + + def _normalize_eos_token_ids(value: Any) -> list[int]: if isinstance(value, int) and not isinstance(value, bool): return [value] @@ -428,6 +437,7 @@ def _build_dynamo_preproc( tool_call_parser is None and reasoning_parser is None ), "return_tokens_as_token_ids": request.get("return_tokens_as_token_ids"), + "continuous_usage_stats": _continuous_usage_requested(request), }, "eos_token_ids": _normalize_eos_token_ids(eos_token_ids), "annotations": [], @@ -705,6 +715,7 @@ async def _generate_and_stream( post_proc_total_ms = 0.0 created_ts = int(time.time()) stream_interval = self.stream_interval + continuous_usage = _continuous_usage_requested(request) try: dynamo_stream = await self.routed_engine.generate( @@ -771,7 +782,11 @@ async def _generate_and_stream( # Flush on finish or when we've accumulated enough tokens. # First chunk flushes immediately (si=1) to minimize TTFT. flush_threshold = 1 if first_chunk else stream_interval - if finish_reason or len(pending_token_ids) >= flush_threshold: + if ( + finish_reason + or len(pending_token_ids) >= flush_threshold + or (continuous_usage and pending_usage and not pending_token_ids) + ): usage_for_metrics = pending_usage mapped_response = { "token_ids": pending_token_ids, @@ -791,10 +806,10 @@ async def _generate_and_stream( token_count += len(pending_token_ids) envelope: dict[str, Any] = {"_dynamo_annotated": True} - if choice: + if choice or (continuous_usage and pending_usage): dynamo_out: dict[str, Any] = { "id": request_id, - "choices": [choice], + "choices": [choice] if choice else [], "created": created_ts, "model": request["model"], "object": "chat.completion.chunk", @@ -814,7 +829,7 @@ async def _generate_and_stream( pending_usage["completion_tokens_details"] = { "reasoning_tokens": reasoning_tokens } - if pending_usage: + if pending_usage and (finish_reason or continuous_usage): dynamo_out["usage"] = pending_usage pending_usage = None response_nvext: dict[str, Any] = {} @@ -827,15 +842,9 @@ async def _generate_and_stream( ): response_nvext["engine_data"] = engine_data - # Promote SGLang's cached_tokens_details to the - # sglext carrier on the finish chunk. The decode - # handler only populates engine_data.cached_tokens_details - # on the terminal chunk, so this naturally fires once. - # Routed as an internal nvext key that the Rust - # serializer lifts to a top-level `sglext` field, - # matching sglang's native protocol — independent of - # the client's engine_data opt-in. - if finish_reason: + # Preserve the engine's cache breakdown alongside running + # usage, including parser-buffered usage-only chunks. + if finish_reason or continuous_usage: cached_tokens_details = ( _cached_tokens_details_from_engine_data(engine_data) ) @@ -874,6 +883,10 @@ async def _generate_and_stream( for key, value in dynamo_out.items() if key not in ("usage", "nvext") } + if continuous_usage and "usage" in dynamo_out: + payload["usage"] = dynamo_out["usage"] + if "nvext" in dynamo_out: + payload["nvext"] = dynamo_out["nvext"] payload["choices"] = [{**choice, "finish_reason": None}] # The logprobs payload rides with the tool-call # payload only; duplicating it onto the pure finish diff --git a/dingo/frontend/tests/test_sglang_processor_metrics_unit.py b/dingo/frontend/tests/test_sglang_processor_metrics_unit.py index 0bfc9837e4a9..1807fc68922b 100644 --- a/dingo/frontend/tests/test_sglang_processor_metrics_unit.py +++ b/dingo/frontend/tests/test_sglang_processor_metrics_unit.py @@ -233,8 +233,13 @@ async def collect(): @pytest.mark.parametrize("finish_reason", ["tool_calls", "length"]) @pytest.mark.parametrize("tool_count", [1, 2]) @pytest.mark.parametrize("stream_interval", [1, 20]) +@pytest.mark.parametrize("continuous", [False, True]) def test_terminal_tool_payload_precedes_finish( - module_stubs: Any, finish_reason: str, tool_count: int, stream_interval: int + module_stubs: Any, + finish_reason: str, + tool_count: int, + stream_interval: int, + continuous: bool, ) -> None: module = _load_processor_module(module_stubs) tool_calls = [ @@ -287,7 +292,15 @@ async def collect() -> list[dict[str, Any]]: item async for item in processor._generate_and_stream( "req-tools", - {"model": "glm-5.3", "nvext": {"extra_fields": ["stop_reason"]}}, + { + "model": "glm-5.3", + "stream": True, + "stream_options": { + "include_usage": True, + "continuous_usage_stats": continuous, + }, + "nvext": {"extra_fields": ["stop_reason"]}, + }, {}, list(range(10)), Post(), @@ -299,7 +312,11 @@ async def collect() -> list[dict[str, Any]]: data, terminal = [item["data"] for item in items] assert data["choices"] == [{**choice, "finish_reason": None}] assert terminal["choices"] == [{**choice, "delta": {}}] - assert "usage" not in data and "nvext" not in data + if continuous: + assert data["usage"] == terminal["usage"] + assert data["nvext"] == terminal["nvext"] + else: + assert "usage" not in data and "nvext" not in data assert terminal["usage"] == { **usage, "completion_tokens_details": {"reasoning_tokens": 1}, @@ -323,3 +340,198 @@ async def collect() -> list[dict[str, Any]]: assert len(metrics) == 1 assert metrics[0]["chunk_tokens"] == metrics[0]["output_tokens"] == 3 assert items[0]["event"] == "llm_metrics" + + +class _ByteTokenizer: + def decode(self, ids, **kwargs): + return bytes(ids).decode("utf-8", errors="replace") + + +def _usage_chunk(ids, count, *, finish=None, cached=4): + usage = { + "prompt_tokens": 10, + "completion_tokens": count, + "total_tokens": 10 + count, + "prompt_tokens_details": ( + {"cached_tokens": cached} if cached is not None else None + ), + } + return { + "token_ids": ids, + "finish_reason": finish, + "completion_usage": usage, + } + + +def _usage_processor(module, chunks, *, interval=1): + processor = module.SglangProcessor( + tokenizer=_ByteTokenizer(), + routed_engine=FakeRoutedEngine(items=chunks), + tool_call_parser_name=None, + reasoning_parser_name=None, + eos_token_ids=None, + ) + processor.stream_interval = interval + post = module.SglangStreamingPostProcessor( + tokenizer=processor.tokenizer, + tool_call_parser=None, + reasoning_parser=None, + ) + return processor, post + + +def _usage_request(options=None, *, stream=True): + request = {"model": "test-model", "stream": stream} + if options is not None: + request["stream_options"] = options + return request + + +_CONTINUOUS_USAGE = {"include_usage": True, "continuous_usage_stats": True} + + +@pytest.mark.parametrize( + "options,expected_counts", + [ + (None, [4]), + ({"include_usage": True}, [4]), + ({"continuous_usage_stats": True}, [4]), + (_CONTINUOUS_USAGE, [1, 2, 3, 4]), + ], +) +def test_usage_survives_buffered_utf8_without_changing_text( + module_stubs, options, expected_counts +): + module = _load_processor_module(module_stubs) + chunks = [ + _usage_chunk([byte], index, finish="stop" if index == 4 else None) + for index, byte in enumerate("你!".encode("utf-8"), start=1) + ] + chunks[0]["engine_data"] = { + "cached_tokens_details": {"device": 2, "host": 2} + } + processor, post = _usage_processor(module, chunks) + + async def collect(): + return [ + item + async for item in processor._generate_and_stream( + "req-buffered", _usage_request(options), {}, list(range(10)), post + ) + ] + + events = asyncio.run(collect()) + data = [event["data"] for event in events if "data" in event] + assert "".join( + choice["delta"].get("content", "") + for item in data + for choice in item["choices"] + ) == "你!" + usages = [item["usage"] for item in data if "usage" in item] + assert [usage["completion_tokens"] for usage in usages] == expected_counts + assert [usage["total_tokens"] for usage in usages] == [ + 10 + count for count in expected_counts + ] + assert all(usage["prompt_tokens_details"]["cached_tokens"] == 4 for usage in usages) + if options == _CONTINUOUS_USAGE: + assert data[0]["choices"] == data[1]["choices"] == [] + assert data[0]["nvext"][module._INTERNAL_SGLEXT_KEY] == { + "cached_tokens_details": {"device": 2, "host": 2} + } + assert data[-1]["choices"][0]["finish_reason"] == "stop" + + +def test_running_usage_preserves_batching_and_engine_terminal_count(module_stubs): + module = _load_processor_module(module_stubs) + chunks = [ + _usage_chunk([byte], index) + for index, byte in enumerate(b"ABCDE", start=1) + ] + # The engine counted a hidden stop token; do not replace its authoritative + # completion count with the number of visible token IDs. + chunks.append(_usage_chunk([], 6, finish="stop")) + processor, post = _usage_processor(module, chunks, interval=3) + + async def collect(): + return [ + item + async for item in processor._generate_and_stream( + "req-batched", + _usage_request(_CONTINUOUS_USAGE), + {}, + list(range(10)), + post, + ) + ] + + events = asyncio.run(collect()) + data = [event["data"] for event in events if "data" in event] + assert [ + item["choices"][0]["delta"].get("content", "") for item in data + ] == ["A", "BCD", "E"] + assert [item["usage"]["completion_tokens"] for item in data] == [1, 4, 6] + assert [item["usage"]["total_tokens"] for item in data] == [11, 14, 16] + # Metrics describe token delivery, not a sum of cumulative usage snapshots. + assert sum(json.loads(event["comment"][0])["chunk_tokens"] for event in events) == 5 + + +@pytest.mark.parametrize("cached", [0, 4, None]) +def test_first_content_usage_remains_available_without_draining_stream( + module_stubs, cached +): + module = _load_processor_module(module_stubs) + chunks = [ + _usage_chunk([65], 1, cached=cached), + _usage_chunk([66], 2, finish="stop", cached=cached), + ] + processor, post = _usage_processor(module, chunks) + + async def disconnect(): + stream = processor._generate_and_stream( + "req-disconnect", + _usage_request(_CONTINUOUS_USAGE), + {}, + list(range(10)), + post, + ) + first = await anext(stream) + await stream.aclose() + return first["data"] + + first = asyncio.run(disconnect()) + assert first["choices"][0]["delta"]["content"] == "A" + assert first["choices"][0]["finish_reason"] is None + assert first["usage"] == { + "prompt_tokens": 10, + "completion_tokens": 1, + "total_tokens": 11, + "prompt_tokens_details": ( + {"cached_tokens": cached} if cached is not None else None + ), + } + + +def test_empty_token_report_emits_usage_without_waiting_for_finish(module_stubs): + module = _load_processor_module(module_stubs) + processor, post = _usage_processor( + module, + [_usage_chunk([], 1), _usage_chunk([65], 2, finish="stop")], + interval=8, + ) + + async def first_event(): + stream = processor._generate_and_stream( + "req-no-visible-token", + _usage_request(_CONTINUOUS_USAGE), + {}, + list(range(10)), + post, + ) + first = await anext(stream) + await stream.aclose() + return first + + event = asyncio.run(first_event()) + assert event["data"]["choices"] == [] + assert event["data"]["usage"]["completion_tokens"] == 1 + assert json.loads(event["comment"][0])["chunk_tokens"] == 0 diff --git a/dingo/sglang/request_handlers/llm/decode_handler.py b/dingo/sglang/request_handlers/llm/decode_handler.py index 2e80909d8d6d..a2ece1ee64be 100644 --- a/dingo/sglang/request_handlers/llm/decode_handler.py +++ b/dingo/sglang/request_handlers/llm/decode_handler.py @@ -486,6 +486,7 @@ async def generate( return_tokens_as_token_ids = bool( output_options.get("return_tokens_as_token_ids") ) + continuous_usage_stats = output_options.get("continuous_usage_stats", False) user_stop_token_ids = _user_stop_token_ids(request) suppressed_stop_token_ids = _suppressed_stop_token_ids(request) @@ -547,6 +548,7 @@ async def generate( user_stop_token_ids=user_stop_token_ids, suppressed_stop_token_ids=suppressed_stop_token_ids, metadata_uploader=metadata_uploader, + continuous_usage_stats=continuous_usage_stats, ): yield out else: @@ -617,6 +619,7 @@ async def generate( user_stop_token_ids=user_stop_token_ids, suppressed_stop_token_ids=suppressed_stop_token_ids, metadata_uploader=metadata_uploader, + continuous_usage_stats=continuous_usage_stats, ): yield out else: @@ -637,6 +640,7 @@ async def _process_token_stream( user_stop_token_ids: set[int] | None = None, suppressed_stop_token_ids: set[int] | None = None, metadata_uploader: MetadataUploader | None = None, + continuous_usage_stats: bool = False, ) -> AsyncGenerator[Dict[str, Any], None]: """Process token-based stream output. @@ -646,6 +650,8 @@ async def _process_token_stream( Args: stream_source: Async generator from engine.async_generate. context: Context object for cancellation handling. + continuous_usage_stats: Emit engine cumulative usage on every update, + including updates whose tokens are buffered as possible stops. Yields: Dict with token_ids and optional finish_reason. @@ -660,13 +666,13 @@ async def _process_token_stream( pending_stop_tokens_per_choice: dict[int, list[int]] = {} pending_log_probs_per_choice: dict[int, list[Any]] = {} pending_top_logprobs_per_choice: dict[int, list[Any]] = {} - # SGLang reports cached_tokens_details on each chunk's meta_info, but - # the frontend only emits it to the client on the terminal chunk. Track - # the latest value per choice (n>1 interleaves choices) and fold it - # into engine_data on the finish chunk so the processor can promote it - # to the sglext carrier. + # Keep the latest cache breakdown per choice for the terminal chunk, + # and expose it on intermediate chunks only in continuous-usage mode. cached_tokens_details_per_choice: dict[int, dict[str, Any]] = {} - async with self._cancellation_monitor(request_id_future, context): + cached_tokens_per_choice: dict[int, int] = {} + async with self._cancellation_monitor( + request_id_future, context + ) as cancellation_task: async for res in stream_source: meta_info = res.get("meta_info", {}) # Extract SGLang request ID from the first response and set the future @@ -676,22 +682,31 @@ async def _process_token_stream( request_id_future.set_result(sglang_request_id) logging.debug(f"New SGLang Request ID: {sglang_request_id}") - # Check cancellation before yielding to allow proper cleanup. - # This lets SGLang proceed to the second token generation, which will - # async context switch and allow the abort monitor to signal cancellation. - # The loop should exit by itself when context.is_stopped() returns True. + # Do not drain more engine updates or manufacture a terminal + # usage event after cancellation. + if context.is_stopped(): + if cancellation_task is not None and request_id_future.done(): + await cancellation_task + break + # SGLang omits index for non-n/legacy chunks; treat those as # choice 0 while preserving explicit indices for n>1. output_idx = res.get("index") or 0 # Capture SGLang's per-dimension cache hit breakdown. SGLang # populates this on every chunk once the radix-tree lookup is - # known; keep the latest per choice and emit it on the finish - # chunk below. Requested via return_cached_tokens_details - # (sglang patch forces it on for all requests). + # known; keep the latest per choice for usage-bearing chunks. + # Requested via return_cached_tokens_details (sglang patch + # forces it on for all requests). cached_tokens_details = meta_info.get("cached_tokens_details") if isinstance(cached_tokens_details, dict): cached_tokens_details_per_choice[output_idx] = cached_tokens_details + cached_tokens = meta_info.get("cached_tokens") + if continuous_usage_stats: + if cached_tokens is not None: + cached_tokens_per_choice[output_idx] = cached_tokens + else: + cached_tokens = cached_tokens_per_choice.get(output_idx) out: dict[str, Any] = {"index": output_idx} finish_reason = meta_info["finish_reason"] @@ -705,6 +720,27 @@ async def _process_token_stream( if stop_reason is not None: out["stop_reason"] = stop_reason + # Token IDs are deltas, but engine usage is already cumulative. + # Missing intermediate counters must not become invented zeros. + has_usage = ( + meta_info.get("prompt_tokens") is not None + and meta_info.get("completion_tokens") is not None + ) + if finish_reason or (continuous_usage_stats and has_usage): + input_tokens = meta_info["prompt_tokens"] + completion_tokens = meta_info["completion_tokens"] + prefill_prompt_tokens_details = None + if cached_tokens is not None and ( + continuous_usage_stats or cached_tokens > 0 + ): + prefill_prompt_tokens_details = {"cached_tokens": cached_tokens} + out["completion_usage"] = { + "prompt_tokens": input_tokens, + "completion_tokens": completion_tokens, + "total_tokens": input_tokens + completion_tokens, + "prompt_tokens_details": prefill_prompt_tokens_details, + } + # With stream_output=True, output_ids contains only new tokens (disjoint). raw_output_ids = list(res.get("output_ids", [])) log_probs = None @@ -769,11 +805,13 @@ async def _process_token_stream( ] top_logprobs = top_logprobs[: -len(pending_stop_tokens)] - # Empty, non-final chunks can happen while an ambiguous stop - # suffix is buffered. Wait until it matches or becomes visible. - if not output_ids and not finish_reason: - if context.is_stopped(): - break + # Preserve real usage even when a possible stop suffix hides + # every token. Legacy streams still wait for visible output. + if ( + not output_ids + and not finish_reason + and "completion_usage" not in out + ): continue out["token_ids"] = output_ids @@ -786,25 +824,17 @@ async def _process_token_stream( engine_data: Dict[str, Any] = {} if routed_experts is not None and metadata_uploader is None: engine_data["routed_experts"] = routed_experts - if finish_reason and output_idx in cached_tokens_details_per_choice: - engine_data["cached_tokens_details"] = ( + if ( + finish_reason or continuous_usage_stats + ) and output_idx in cached_tokens_details_per_choice: + out_cache_details = cached_tokens_details_per_choice[output_idx] + engine_data["cached_tokens_details"] = out_cache_details + if finish_reason: cached_tokens_details_per_choice.pop(output_idx) - ) if engine_data: out["engine_data"] = engine_data if finish_reason: - input_tokens = meta_info["prompt_tokens"] - completion_tokens = meta_info["completion_tokens"] - cached_tokens = meta_info["cached_tokens"] - prefill_prompt_tokens_details = None - if cached_tokens is not None and cached_tokens > 0: - prefill_prompt_tokens_details = {"cached_tokens": cached_tokens} - out["completion_usage"] = { - "prompt_tokens": input_tokens, - "completion_tokens": completion_tokens, - "total_tokens": input_tokens + completion_tokens, - "prompt_tokens_details": prefill_prompt_tokens_details, - } + cached_tokens_per_choice.pop(output_idx, None) if metadata_uploader is not None: try: await metadata_uploader.upload_choice(output_idx, meta_info) @@ -812,8 +842,11 @@ async def _process_token_stream( meta_info.clear() elif metadata_uploader is not None: meta_info.clear() - if not context.is_stopped(): - yield out + if context.is_stopped(): + if cancellation_task is not None and request_id_future.done(): + await cancellation_task + break + yield out async def _process_text_stream( self, diff --git a/dingo/sglang/tests/test_sglang_decode_handler.py b/dingo/sglang/tests/test_sglang_decode_handler.py index 4b321fcd8478..0299f15bf4c1 100644 --- a/dingo/sglang/tests/test_sglang_decode_handler.py +++ b/dingo/sglang/tests/test_sglang_decode_handler.py @@ -1,12 +1,14 @@ # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 +import asyncio import json from contextlib import asynccontextmanager from types import SimpleNamespace import pytest +from dingo.common.constants import DisaggregationMode from dingo.common.metadata_upload import MetadataUploader from dingo.sglang.request_handlers.llm.decode_handler import ( DecodeWorkerHandler, @@ -584,6 +586,313 @@ async def test_metadata_upload_normalizes_numpy_values(tmp_path): assert uploaded_array["data"] == expected.tobytes() +def _usage_update( + token_ids, + completion_tokens, + *, + index=0, + cached_tokens=0, + finish_reason=None, + cached_tokens_details=None, +): + meta_info = { + "id": "request-1", + "finish_reason": finish_reason, + "prompt_tokens": 8, + "completion_tokens": completion_tokens, + "cached_tokens": cached_tokens, + } + if cached_tokens_details is not None: + meta_info["cached_tokens_details"] = cached_tokens_details + return {"index": index, "output_ids": token_ids, "meta_info": meta_info} + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "serving_mode", [DisaggregationMode.AGGREGATED, DisaggregationMode.DECODE] +) +@pytest.mark.parametrize("continuous_usage_stats", [None, False, True]) +@pytest.mark.parametrize("cached_tokens", [0, 6, None]) +async def test_generate_emits_cumulative_usage_only_when_requested( + serving_mode, continuous_usage_stats, cached_tokens +): + handler = _new_decode_handler() + handler.serving_mode = serving_mode + handler.enable_trace = False + handler._routed_experts_kwargs = {} + handler._enable_frontend_decoding = False + handler._mm_hashes_supported = False + handler._get_input_param = lambda request: {"input_ids": request["token_ids"]} + handler._resolve_lora = lambda request: None + handler._session_kwargs = lambda request: {} + handler._priority_kwargs = lambda priority: {} + cache_details = {"device": cached_tokens} if cached_tokens is not None else None + + async def generate(**kwargs): + return _stream( + [ + _usage_update( + [101], + 1, + cached_tokens=cached_tokens, + cached_tokens_details=cache_details, + ), + _usage_update([102, 103, 104], 4, cached_tokens=cached_tokens), + _usage_update( + [], + 4, + cached_tokens=cached_tokens, + finish_reason={"type": "stop"}, + ), + ] + ) + + handler.engine = SimpleNamespace(async_generate=generate) + request = { + "token_ids": list(range(8)), + "stop_conditions": {"max_tokens": 4}, + "bootstrap_info": { + "bootstrap_host": "localhost", + "bootstrap_port": 1234, + "bootstrap_room": 42, + }, + } + if continuous_usage_stats is not None: + request["output_options"] = { + "continuous_usage_stats": continuous_usage_stats, + } + context = _Context() + context.id = lambda: "request-1" + context.trace_id = "trace-1" + + chunks = await _collect(handler.generate(request, context)) + + assert [chunk["token_ids"] for chunk in chunks] == [[101], [102, 103, 104], []] + assert ["finish_reason" in chunk for chunk in chunks] == [False, False, True] + assert chunks[-1]["finish_reason"] == "stop" + expected_details = ( + {"cached_tokens": cached_tokens} + if cached_tokens is not None and (continuous_usage_stats or cached_tokens > 0) + else None + ) + for chunk, count in zip(chunks, [1, 4, 4]): + if continuous_usage_stats or "finish_reason" in chunk: + assert chunk["completion_usage"] == { + "prompt_tokens": 8, + "completion_tokens": count, + "total_tokens": 8 + count, + "prompt_tokens_details": expected_details, + } + if cache_details is not None: + assert chunk["engine_data"]["cached_tokens_details"] == cache_details + else: + assert "engine_data" not in chunk + else: + assert "completion_usage" not in chunk + assert "engine_data" not in chunk + + +@pytest.mark.asyncio +@pytest.mark.parametrize("continuous_usage_stats", [False, True]) +async def test_process_token_stream_preserves_empty_usage_updates(continuous_usage_stats): + handler = _new_decode_handler() + missing_usage = { + "output_ids": [], + "meta_info": {"id": "request-1", "finish_reason": None}, + } + partial_usage = { + "output_ids": [101], + "meta_info": { + "id": "request-1", + "finish_reason": None, + "prompt_tokens": 8, + }, + } + chunks = await _collect( + handler._process_token_stream( + _stream( + [ + missing_usage, + _usage_update([], 0), + partial_usage, + _usage_update([], 1, finish_reason={"type": "stop"}), + ] + ), + _Context(), + continuous_usage_stats=continuous_usage_stats, + ) + ) + + assert [chunk["token_ids"] for chunk in chunks] == ( + [[], [101], []] if continuous_usage_stats else [[101], []] + ) + assert "completion_usage" not in chunks[-2] + if continuous_usage_stats: + assert chunks[0]["completion_usage"]["completion_tokens"] == 0 + assert chunks[0]["completion_usage"]["total_tokens"] == 8 + assert "finish_reason" not in chunks[0] + assert chunks[-1]["completion_usage"]["completion_tokens"] == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("continuous_usage_stats", [False, True]) +async def test_process_token_stream_keeps_buffered_usage_and_cache_per_choice( + continuous_usage_stats, +): + handler = _new_decode_handler() + chunks = await _collect( + handler._process_token_stream( + _stream( + [ + _usage_update( + [128001], + 1, + cached_tokens=6, + cached_tokens_details={"device": 6}, + ), + _usage_update( + [201], + 1, + index=1, + cached_tokens_details={"device": 0}, + ), + _usage_update( + [128009], + 2, + cached_tokens=6, + finish_reason={ + "type": "stop", + "matched": [128001, 128009], + }, + ), + _usage_update( + [202, 203], + 3, + index=1, + cached_tokens_details={"device": 0, "host": 0}, + ), + _usage_update( + [], 3, index=1, finish_reason={"type": "length"} + ), + ] + ), + _Context(), + suppressed_stop_token_ids={128001, 128009}, + continuous_usage_stats=continuous_usage_stats, + ) + ) + + assert [chunk["index"] for chunk in chunks] == ( + [0, 1, 0, 1, 1] if continuous_usage_stats else [1, 0, 1, 1] + ) + assert [chunk["token_ids"] for chunk in chunks] == ( + [[], [201], [], [202, 203], []] + if continuous_usage_stats + else [[201], [], [202, 203], []] + ) + usage_chunks = [chunk for chunk in chunks if "completion_usage" in chunk] + assert [chunk["completion_usage"]["completion_tokens"] for chunk in usage_chunks] == ( + [1, 1, 2, 3, 3] if continuous_usage_stats else [2, 3] + ) + for chunk in usage_chunks: + cached = 6 if chunk["index"] == 0 else 0 + details = chunk["completion_usage"]["prompt_tokens_details"] + assert details == ( + {"cached_tokens": cached} if continuous_usage_stats or cached else None + ) + assert chunk["engine_data"]["cached_tokens_details"]["device"] == cached + assert chunks[-1]["engine_data"]["cached_tokens_details"] == {"device": 0, "host": 0} + assert chunks[-1]["finish_reason"] == "length" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("continuous_usage_stats", [False, True]) +@pytest.mark.parametrize("cancelled_token_ids", [[], [128001], [102]]) +async def test_process_token_stream_cancellation_aborts_without_terminal_usage( + continuous_usage_stats, cancelled_token_ids +): + handler = _new_decode_handler() + # Exercise the real cancellation monitor, not the no-op fixture. + del handler._cancellation_monitor + handler.shutdown_event = None + aborted = [] + handler.engine = SimpleNamespace( + tokenizer_manager=SimpleNamespace( + abort_request=lambda **kwargs: aborted.append(kwargs) + ) + ) + cancellation = asyncio.get_running_loop().create_future() + context = SimpleNamespace( + id=lambda: "request-1", + is_stopped=lambda: cancellation.done(), + async_killed_or_stopped=lambda: cancellation, + ) + + async def cancelled_stream(): + yield _usage_update([101], 1) + cancellation.set_result(None) + yield _usage_update(cancelled_token_ids, 2) + pytest.fail("Cancelled generation must not drain further engine updates") + + chunks = await asyncio.wait_for( + _collect( + handler._process_token_stream( + cancelled_stream(), + context, + suppressed_stop_token_ids={128001}, + continuous_usage_stats=continuous_usage_stats, + ) + ), + timeout=1, + ) + + assert len(chunks) == 1 + assert chunks[0]["token_ids"] == [101] + assert "finish_reason" not in chunks[0] + assert ("completion_usage" in chunks[0]) is continuous_usage_stats + if continuous_usage_stats: + assert chunks[0]["completion_usage"]["completion_tokens"] == 1 + assert aborted == [{"rid": "request-1", "abort_all": False}] + + +@pytest.mark.asyncio +async def test_process_token_stream_retains_sparse_cache_usage_before_metadata_cleanup( + tmp_path, +): + handler = _new_decode_handler() + uploader = MetadataUploader(url=(tmp_path / "usage-metadata").as_uri()) + updates = [ + _usage_update([101], 1, cached_tokens=None), + _usage_update([102], 2, cached_tokens=6), + _usage_update([103], 3, cached_tokens=None), + _usage_update([104], 4, cached_tokens=0), + _usage_update([], 4, cached_tokens=None, finish_reason={"type": "stop"}), + ] + chunks = await _collect( + handler._process_token_stream( + _stream(updates), + _Context(), + metadata_uploader=uploader, + continuous_usage_stats=True, + ) + ) + assert [ + chunk["completion_usage"]["prompt_tokens_details"] for chunk in chunks + ] == [ + None, + {"cached_tokens": 6}, + {"cached_tokens": 6}, + {"cached_tokens": 0}, + {"cached_tokens": 0}, + ] + assert [ + chunk["completion_usage"]["completion_tokens"] for chunk in chunks + ] == [1, 2, 3, 4, 4] + assert all(update["meta_info"] == {} for update in updates) + payload = _read_zstd_payload(tmp_path / "usage-metadata/choice_0.msgpack.zst") + assert payload["metadata"]["completion_tokens"] == 4 + + @pytest.mark.asyncio async def test_process_token_stream_tracks_logprobs_per_choice_index(): handler = _new_decode_handler() diff --git a/docs/backends/sglang/README.md b/docs/backends/sglang/README.md index 2480bc3eb28b..48c61d33e8d0 100644 --- a/docs/backends/sglang/README.md +++ b/docs/backends/sglang/README.md @@ -162,6 +162,49 @@ curl localhost:8000/v1/chat/completions \ "max_tokens": 30 }' ``` + +### DingoRouter continuous streaming usage + +The DingoRouter custom frontend (`--dyn-chat-processor sglang`) can request +engine-reported cumulative usage from the SGLang token worker, for both +aggregated serving and the disaggregated decode handler: + +```json +{ + "stream": true, + "stream_options": { + "include_usage": true, + "continuous_usage_stats": true + } +} +``` + +Both options must be true to enable intermediate usage. Other requests retain +the existing terminal/non-streaming behavior. This path requires the updated +Python frontend and worker **and** rebuilt `ai-dingo-runtime` bindings: the +internal `OutputOptions.continuous_usage_stats` field must survive every Rust +request-serialization boundary. Installing SMG is not required. + +Usage snapshots come from the engine's `meta_info`, not from re-tokenizing +visible text or adding together previous snapshots. `prompt_tokens_details.cached_tokens` +preserves an engine-reported zero; an unavailable cache count stays absent/null. +Previously reported cache metadata is retained when a later update omits it. +The SGLang cache breakdown, when provided, is exposed through `sglext`. + +If the decoder/parser withholds text (for example an incomplete UTF-8 character +or tool-call syntax), a usage-only chunk with `choices: []` may arrive before +visible text. Consumers must accept these chunks without treating them as end +of stream. `DYN_SGLANG_STREAM_INTERVAL` still controls token batching: each emitted +usage snapshot is cumulative, not a guarantee of one SSE event per GPU decode step. + +For disconnect billing, retain the last valid usage snapshot and settle once; +never sum snapshots or treat repeated terminal/tool chunks as additional usage. +No usage is invented before the engine supplies counters or after cancellation. +This does not provide final engine consumption after a disconnected stream, nor +does it implement billing settlement in Nexus. The custom frontend still accepts +only `n=1`. Separate native-tokenizer/text-mode and unified backend entry points +are outside this custom-processor feature. + ### Disaggregated Serving Launch a disaggregated Qwen3-0.6B deployment (smallest model, useful for plumbing validation): diff --git a/lib/llm/src/protocols/common.rs b/lib/llm/src/protocols/common.rs index 49dcd143e52a..5c4f6e6cb5e2 100644 --- a/lib/llm/src/protocols/common.rs +++ b/lib/llm/src/protocols/common.rs @@ -557,6 +557,11 @@ pub struct OutputOptions { /// When true, logprob token fields are returned as "token_id:" /// instead of decoded text. pub return_tokens_as_token_ids: Option, + + /// Request cumulative engine usage on intermediate responses. + /// Used by custom processors that assemble their own OpenAI stream. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub continuous_usage_stats: bool, } // Struct for log probability information diff --git a/lib/llm/src/protocols/openai.rs b/lib/llm/src/protocols/openai.rs index fa9c6c29571f..dffb6da626cc 100644 --- a/lib/llm/src/protocols/openai.rs +++ b/lib/llm/src/protocols/openai.rs @@ -229,6 +229,7 @@ impl OutputOptionsProvider for T { skip_special_tokens, formatted_prompt, return_tokens_as_token_ids, + continuous_usage_stats: false, }) } } From 0ded29c1ce3e9b02379b3407e1a3a64203568652 Mon Sep 17 00:00:00 2001 From: githubgxll <1094462054@qq.com> Date: Wed, 7 Oct 2026 20:46:32 +0800 Subject: [PATCH 2/3] build: default to SGLang v0.5.21 runtime image --- .github/dingo-images.json | 4 ++-- container/context.yaml | 16 ++++++---------- 2 files changed, 8 insertions(+), 12 deletions(-) diff --git a/.github/dingo-images.json b/.github/dingo-images.json index b43d95bb8e01..0de73b9a663b 100644 --- a/.github/dingo-images.json +++ b/.github/dingo-images.json @@ -21,12 +21,12 @@ }, { "framework": "vllm", - "enabled": true, + "enabled": false, "repository": "ai-dingo-vllm" }, { "framework": "sglang", - "enabled": false, + "enabled": true, "repository": "ai-dingo-sglang", "docker_target": "pre_runtime" } diff --git a/container/context.yaml b/container/context.yaml index ff4f9d0fe64b..a071cf67ca9f 100644 --- a/container/context.yaml +++ b/container/context.yaml @@ -96,16 +96,12 @@ sglang: base_image: nvcr.io/nvidia/cuda-dl-base runtime_image: lmsysorg/sglang base_image_tag: 25.11-cuda13.0-devel-ubuntu24.04 - runtime_image_tag: glm-5.3-flash - # Baseline is the TRUE base of the third-party lmsysorg/sglang image, - # nvidia/cuda:13.0.1-cudnn-devel-ubuntu24.04 (Docker Hub; CUDA_VERSION=13.0.1, - # cuDNN 9.13.0.50, ubuntu24.04), NOT a self-baseline of lmsysorg/sglang. - # lmsysorg/sglang is built FROM this nvidia/cuda cudnn-devel base, so this - # attributes everything sglang adds over the clean NVIDIA CUDA+cuDNN floor - # (framework, python stack, system libs) rather than hiding it. Per-arch stem; - # the licenses stage appends -${TARGETARCH}.cdx.json. (sglang ships no system - # GPL ffmpeg/codec dpkgs; the GPL imageio-ffmpeg pip binary is replaced by an - # in-tree LGPL ffmpeg in sglang_runtime.Dockerfile — no dpkg purge needed.) + runtime_image_tag: v0.5.21 + # Historical CUDA 13.0.1 compliance corpus; not refreshed for this runtime + # update. Upstream v0.5.21 uses CUDA 13.0.3, so refresh this baseline before + # publishing compliance artifacts for the new image. Dingo image CI uses + # the pre_runtime target and does not publish SBOM/provenance artifacts. + # Per-arch stem; the licenses stage appends -${TARGETARCH}.cdx.json. baseline_sbom: cuda@8b2705ea xpu: base_image: intel/deep-learning-essentials From 717e5b7fd2859a88a39d8f4b9119341c5fd12f8a Mon Sep 17 00:00:00 2001 From: githubgxll <1094462054@qq.com> Date: Sun, 11 Oct 2026 09:56:13 +0800 Subject: [PATCH 3/3] ci: retry image build (docker hub transient EOF pulling lmsysorg/sglang:v0.5.21)