diff --git a/dingo/common/backend/logprobs.py b/dingo/common/backend/logprobs.py index ee7dc5926682..6bab1cf2a68f 100644 --- a/dingo/common/backend/logprobs.py +++ b/dingo/common/backend/logprobs.py @@ -227,22 +227,28 @@ def extract_prompt_logprobs_from_sglang_meta( _SGLANG_TOP_LOGPROBS_UNSUPPORTED_MSG = ( - "Dynamo's SGLang backend does not currently support logprobs >= 1 due to " - "an O(N) per-position detokenization in the upstream sglang tokenizer " - "manager. Use logprobs=0 for chosen-token logprobs, or set " - "DYN_SGL_ALLOW_TOP_LOGPROBS=1 to override at your own risk. " - "Track the upstream fix at https://github.com/sgl-project/sglang/pull/24447." + "SGLang top-k logprobs are disabled by DYN_SGL_ALLOW_TOP_LOGPROBS=0. " + "Set DYN_SGL_ALLOW_TOP_LOGPROBS=1 to enable them. SGLang versions without " + "batched top-token detokenization may incur extra latency for long outputs. " + "See the upstream optimization proposal at " + "https://github.com/sgl-project/sglang/pull/24447." ) DYN_SGL_ALLOW_TOP_LOGPROBS_ENV = "DYN_SGL_ALLOW_TOP_LOGPROBS" def sglang_top_logprobs_allowed() -> bool: - """Read the ``DYN_SGL_ALLOW_TOP_LOGPROBS`` env-var gate.""" - return os.environ.get(DYN_SGL_ALLOW_TOP_LOGPROBS_ENV, "").lower() not in ( - "", + """Return whether SGLang top-k logprobs are enabled. + + They are enabled by default so valid OpenAI ``logprobs`` requests work. + Set ``DYN_SGL_ALLOW_TOP_LOGPROBS=0`` to restore the opt-out guard on + deployments where SGLang's per-position detokenization cost is a concern. + """ + return os.environ.get(DYN_SGL_ALLOW_TOP_LOGPROBS_ENV, "1").lower() not in ( "0", "false", + "no", + "off", ) @@ -255,10 +261,8 @@ def build_sglang_logprob_kwargs( ``return_logprob`` / ``top_logprobs_num`` / ``logprob_start_len`` kwargs. Raises ``ValueError`` for ``logprobs >= 1`` when - ``allow_top_logprobs`` is ``False``. SGLang's tokenizer manager - detokenizes top-k tokens serially (O(N) per generated token), so - enabling it without a batched detokenize path degrades latency - badly. + ``allow_top_logprobs`` is ``False``. This is an opt-out guard for + deployments concerned about per-position top-token detokenization cost. """ if not output_options: return {} @@ -298,28 +302,44 @@ def extract_from_sglang_meta( num_output_logprobs_so_far: int, *, return_tokens_as_token_ids: bool = False, + incremental: bool = False, ) -> tuple[Optional[list[float]], Optional[list[list[dict[str, Any]]]], int]: """Extract logprobs from SGLang's ``meta_info`` dict. - SGLang's ``output_token_logprobs`` / ``output_top_logprobs`` are - cumulative across stream chunks even though ``output_ids`` is - disjoint — the caller passes the running count to slice the new - entries, and the returned third element is the updated count. + When ``incremental_streaming_output`` is False (the SGLang default), + ``output_token_logprobs`` / ``output_top_logprobs`` are cumulative + across stream chunks even though ``output_ids`` is disjoint — the + caller passes the running count to slice the new entries, and the + returned third element is the updated count. + + When ``incremental_streaming_output`` is True (Dynamo forces this on), + SGLang sends only the new token's logprobs in each chunk — the arrays + are already disjoint, so no slicing is needed. The returned third + element is ``num_output_logprobs_so_far + len(new_entries)``. """ output_token_logprobs = meta_info.get("output_token_logprobs") if not output_token_logprobs: return None, None, num_output_logprobs_so_far - new_logprobs = output_token_logprobs[num_output_logprobs_so_far:] + if incremental: + new_logprobs = output_token_logprobs + new_total = num_output_logprobs_so_far + len(output_token_logprobs) + else: + new_logprobs = output_token_logprobs[num_output_logprobs_so_far:] + new_total = len(output_token_logprobs) + if not new_logprobs: - return None, None, num_output_logprobs_so_far + return None, None, new_total log_probs = [float(entry[0]) for entry in new_logprobs] top_logprobs: Optional[list[list[dict[str, Any]]]] = None output_top = meta_info.get("output_top_logprobs") if output_top: - new_top = output_top[num_output_logprobs_so_far:] + if incremental: + new_top = output_top + else: + new_top = output_top[num_output_logprobs_so_far:] if new_top: top_logprobs = [] for position_entries in new_top: @@ -330,7 +350,9 @@ def extract_from_sglang_meta( for rank_idx, entry in enumerate(position_entries): tok_id = entry[1] token_str = ( - f"token_id:{tok_id}" if return_tokens_as_token_ids else entry[2] + f"token_id:{tok_id}" + if return_tokens_as_token_ids + else entry[2] ) position_list.append( { @@ -342,4 +364,4 @@ def extract_from_sglang_meta( ) top_logprobs.append(position_list) - return log_probs, top_logprobs, len(output_token_logprobs) + return log_probs, top_logprobs, new_total diff --git a/dingo/common/backend/tests/test_logprobs.py b/dingo/common/backend/tests/test_logprobs.py index 736d7ac46034..45bfa00ff797 100644 --- a/dingo/common/backend/tests/test_logprobs.py +++ b/dingo/common/backend/tests/test_logprobs.py @@ -301,13 +301,13 @@ def test_sglang_kwargs_empty_when_no_options(): def test_sglang_kwargs_logprobs_zero_allowed_without_gate(): - # The default gate forbids logprobs >= 1; logprobs=0 always works. + # Chosen-token-only logprobs are available even when top-k is disabled. kwargs = build_sglang_logprob_kwargs({"logprobs": 0}, allow_top_logprobs=False) assert kwargs == {"return_logprob": True, "top_logprobs_num": 0} def test_sglang_kwargs_top_logprobs_rejected_without_gate(): - with pytest.raises(ValueError, match="does not currently support logprobs >= 1"): + with pytest.raises(ValueError, match="disabled by DYN_SGL_ALLOW_TOP_LOGPROBS=0"): build_sglang_logprob_kwargs({"logprobs": 2}, allow_top_logprobs=False) @@ -336,12 +336,16 @@ def test_sglang_kwargs_both_set_picks_max(): def test_sglang_gate_reads_env(monkeypatch): + monkeypatch.delenv(DYN_SGL_ALLOW_TOP_LOGPROBS_ENV, raising=False) + assert sglang_top_logprobs_allowed() is True monkeypatch.setenv(DYN_SGL_ALLOW_TOP_LOGPROBS_ENV, "1") assert sglang_top_logprobs_allowed() is True monkeypatch.setenv(DYN_SGL_ALLOW_TOP_LOGPROBS_ENV, "0") assert sglang_top_logprobs_allowed() is False - monkeypatch.delenv(DYN_SGL_ALLOW_TOP_LOGPROBS_ENV, raising=False) + monkeypatch.setenv(DYN_SGL_ALLOW_TOP_LOGPROBS_ENV, "false") assert sglang_top_logprobs_allowed() is False + monkeypatch.delenv(DYN_SGL_ALLOW_TOP_LOGPROBS_ENV, raising=False) + assert sglang_top_logprobs_allowed() is True # --------------------------------------------------------------------------- @@ -410,6 +414,43 @@ def test_sglang_extract_returns_offset_unchanged_when_no_new_entries(): assert new_total == 1 +def test_sglang_extract_incremental_no_slice(): + """incremental=True: arrays are already disjoint, no slicing.""" + meta = { + "output_token_logprobs": [(-0.5, 10, "x")], + "output_top_logprobs": [[(-0.5, 10, "x"), (-0.8, 11, "y")]], + } + log_probs, top_logprobs, new_total = extract_from_sglang_meta( + meta, 5, incremental=True + ) + assert log_probs == [-0.5] + assert top_logprobs == [ + [ + {"rank": 1, "token_id": 10, "token": "x", "logprob": -0.5}, + {"rank": 2, "token_id": 11, "token": "y", "logprob": -0.8}, + ] + ] + assert new_total == 6 + + +def test_sglang_extract_incremental_empty_array(): + """incremental=True: empty output_token_logprobs returns None.""" + meta = {"output_token_logprobs": []} + log_probs, _, new_total = extract_from_sglang_meta(meta, 3, incremental=True) + assert log_probs is None + assert new_total == 3 + + +def test_sglang_extract_incremental_no_top_logprobs(): + """incremental=True: log_probs without top_logprobs.""" + meta = {"output_token_logprobs": [(-0.1, 1, "a")]} + log_probs, top_logprobs, new_total = extract_from_sglang_meta( + meta, 0, incremental=True + ) + assert log_probs == [-0.1] + assert top_logprobs is None + assert new_total == 1 + # --------------------------------------------------------------------------- # Legacy ↔ unified behavioural parity corner cases. # @@ -656,7 +697,9 @@ def test_parity_sglang_kwargs_rejects_top_logprobs_consistently(): {"prompt_logprobs": 2}, {"logprobs": 0, "prompt_logprobs": 3}, ): - with pytest.raises(ValueError, match="does not currently support"): + with pytest.raises( + ValueError, match="SGLang top-k logprobs are disabled" + ): build_sglang_logprob_kwargs(opts, allow_top_logprobs=False) diff --git a/dingo/frontend/sglang_prepost.py b/dingo/frontend/sglang_prepost.py index c253b7cfcf5a..fa651df726e8 100644 --- a/dingo/frontend/sglang_prepost.py +++ b/dingo/frontend/sglang_prepost.py @@ -735,6 +735,238 @@ def _call_with_optional_parallel_tool_calls( return func(*args) +def _escape_pointer_token(token: str) -> str: + return token.replace("~", "~0").replace("/", "~1") + + +def _unescape_pointer_token(token: str) -> str: + return token.replace("~1", "/").replace("~0", "~") + + +# Keywords whose values are literal instance data, not subschemas (JSON +# Schema spec: const/enum hold instance values, default/examples are +# annotations carrying instance values). Reference rewriting must skip +# them; see _rewrite_def_refs. +_LITERAL_VALUE_KEYWORDS = frozenset({"const", "enum", "default", "examples"}) + +# Keywords whose values are maps from an arbitrary, user-defined name to a +# subschema. The map keys are names, not schema keywords, so the +# literal-value skip must not apply while iterating them: a property named +# "default" still holds a subschema that may contain a ``$ref`` (review +# 20260928 finding 1). +_SCHEMA_MAP_KEYWORDS = frozenset( + { + "properties", + "patternProperties", + "$defs", + "definitions", + "dependentSchemas", + "dependencies", + } +) + + +def _rewrite_def_refs( + value: Any, names: dict[str, str], *, _inside_schema_map: bool = False +) -> None: + """Rewrite ``#/$defs/...`` references in-place according to ``names``. + + Only the leading definition token is remapped; any JSON-pointer suffix + below the definition is preserved. + """ + if isinstance(value, dict): + if not _inside_schema_map: + ref = value.get("$ref") + if isinstance(ref, str) and ref.startswith("#/$defs/"): + pointer = ref[len("#/$defs/") :] + old_token, separator, suffix = pointer.partition("/") + old_name = _unescape_pointer_token(old_token) + new_name = names.get(old_name) + if new_name is not None and new_name != old_name: + value["$ref"] = f"#/$defs/{_escape_pointer_token(new_name)}" + ( + f"/{suffix}" if separator else "" + ) + for key, child in value.items(): + if _inside_schema_map: + # Map keys are user-defined names; every value is a + # subschema and must be traversed whatever it is called. + _rewrite_def_refs(child, names) + elif key in _LITERAL_VALUE_KEYWORDS: + # Literal instance data (const/enum/default/examples), not + # a subschema: a {"$ref": ...} object here is data the + # model must produce and must never be rewritten (review + # 20260924 v2). + continue + elif key in _SCHEMA_MAP_KEYWORDS: + _rewrite_def_refs(child, names, _inside_schema_map=True) + else: + _rewrite_def_refs(child, names) + elif isinstance(value, list): + for child in value: + _rewrite_def_refs(child, names) + + +def _namespace_tool_parameter_defs(tools: list[Any]) -> list[Any]: + """Isolate each tool's parameter ``$defs`` into a unique namespace. + + SGLang's constraint builders collect all tools' ``$defs`` up front and + raise ``ValueError`` when two tools define the same name with different + schemas — before any post-builder processing can run. When two or more + tools carry ``$defs``, deep-copy the tools and rename every definition + to ``{tool_name}__{def_name}``, rewriting that tool's internal + ``#/$defs/...`` references to match. Generated names are reserved + globally up front so they cannot collide with each other (e.g. tool + ``a`` defining ``b__X`` vs tool ``a__b`` defining ``X``). Requests + where at most one tool carries ``$defs`` are returned unchanged: no + cross-tool conflict is possible and original definition names (and + their established behaviour) are preserved. + """ + + def _params_of(tool: Any) -> Any: + fn = getattr(tool, "function", None) + return getattr(fn, "parameters", None) if fn is not None else None + + tools_with_defs = [ + t + for t in tools + if isinstance(_params_of(t), dict) + and isinstance(_params_of(t).get("$defs"), dict) + ] + if len(tools_with_defs) < 2: + return tools + + new_tools = copy.deepcopy(tools) + + # Pass 1: reserve every generated name up front. + taken: set[str] = set() + plans: list[tuple[dict[str, Any], dict[str, str]]] = [] + for tool in new_tools: + params = _params_of(tool) + if not (isinstance(params, dict) and isinstance(params.get("$defs"), dict)): + continue + tool_name = getattr(tool.function, "name", "") or "" + names: dict[str, str] = {} + for old_name in params["$defs"]: + base = f"{tool_name}__{old_name}" + new_name = base + suffix = 2 + while new_name in taken: + new_name = f"{base}_{suffix}" + suffix += 1 + taken.add(new_name) + names[old_name] = new_name + plans.append((params, names)) + + # Pass 2: rewrite references and rename the definition keys. + for params, names in plans: + nested_defs = params.pop("$defs") + _rewrite_def_refs(params, names) + for definition in nested_defs.values(): + _rewrite_def_refs(definition, names) + params["$defs"] = {names[old]: d for old, d in nested_defs.items()} + return new_tools + + +def _hoist_tool_parameter_defs(schema: Any) -> Any: + """Hoist tool parameter ``$defs`` to the guided schema's document root. + + SGLang wraps named/required tool parameters in an outer tool-call schema. + A parameter schema's local ``#/$defs/...`` references then point at the + outer document root, while its ``$defs`` would otherwise remain nested + below ``items.properties.parameters``. Move those definitions to the + outer root and rename on collision. + + Allocation is done in two passes: every final name is reserved up front + — against the existing root ``$defs``, against names allocated earlier + in this pass, and against original names not yet processed — and all + references are rewritten only after every name is known. Definitions + are never deduplicated by dict equality: two definitions with identical + literals can still differ semantically once their own ``$ref`` + dependencies are resolved (review 20260924 finding 3). + """ + if not isinstance(schema, dict): + return schema + + schema = copy.deepcopy(schema) + had_root_defs = "$defs" in schema + root_defs = schema.get("$defs") + if root_defs is None: + root_defs = {} + schema["$defs"] = root_defs + elif not isinstance(root_defs, dict): + return schema + + # Pass 1: collect every parameters node carrying $defs, in visit order. + entries: list[tuple[dict[str, Any], str, dict[str, Any]]] = [] + + def collect_entries(node: Any) -> None: + if isinstance(node, list): + for child in node: + collect_entries(child) + return + if not isinstance(node, dict): + return + + properties = node.get("properties") + if isinstance(properties, dict) and "name" in properties: + parameters = properties.get("parameters") + if isinstance(parameters, dict): + name_schema = properties["name"] + names = ( + name_schema.get("enum", []) + if isinstance(name_schema, dict) + else [] + ) + tool_name = names[0] if isinstance(names, list) and names else "" + nested_defs = parameters.get("$defs") + if isinstance(nested_defs, dict): + entries.append((parameters, tool_name, nested_defs)) + + for keyword in ("items", "anyOf", "oneOf", "allOf", "prefixItems"): + if keyword in node: + collect_entries(node[keyword]) + + collect_entries(schema) + + # Pass 2: allocate a unique final name per definition. A definition + # keeps its original name only when no root definition and no earlier + # allocation already holds it; otherwise it is renamed with a + # ``{tool_name}__`` prefix and a numeric suffix until free. Because + # allocations are sequential and never deduplicated, a renamed + # definition can shadow a not-yet-processed original name only by + # taking it first — the later original is then renamed instead, and + # both keep their own rewritten references. + used: set[str] = set(root_defs) + allocations: list[tuple[dict[str, Any], dict[str, Any], dict[str, str]]] = [] + for parameters, tool_name, nested_defs in entries: + names: dict[str, str] = {} + for old_name in nested_defs: + new_name = old_name + if new_name in used: + base_name = f"{tool_name}__{old_name}" if tool_name else old_name + new_name = base_name + suffix = 2 + while new_name in used: + new_name = f"{base_name}_{suffix}" + suffix += 1 + used.add(new_name) + names[old_name] = new_name + allocations.append((parameters, nested_defs, names)) + + # Pass 3: detach $defs, rewrite references, and write to the root. + for parameters, nested_defs, names in allocations: + parameters.pop("$defs", None) + _rewrite_def_refs(parameters, names) + for definition in nested_defs.values(): + _rewrite_def_refs(definition, names) + for old_name, definition in nested_defs.items(): + root_defs[names[old_name]] = definition + + if not had_root_defs and not root_defs: + schema.pop("$defs", None) + return schema + + def build_tool_call_guided_decoding( request: dict[str, Any], *, @@ -749,6 +981,13 @@ def build_tool_call_guided_decoding( if tool_choice == "none": return None + # Isolate per-tool parameter $defs BEFORE handing tools to SGLang's + # constraint builders: their _get_tool_schema_defs() raises ValueError + # on same-name/different-schema definitions, which would otherwise + # fire before the post-builder hoisting below can run (review 20260924 + # finding 2). At most one tool with $defs passes through unchanged. + sglang_tools = _namespace_tool_parameter_defs(sglang_tools) + parallel_tool_calls = request.get("parallel_tool_calls") constraint: Any = None @@ -788,7 +1027,7 @@ def build_tool_call_guided_decoding( if isinstance(constraint, tuple) and len(constraint) == 2: if constraint[0] == "json_schema": - return {"json": constraint[1]} + return {"json": _hoist_tool_parameter_defs(constraint[1])} if constraint[0] == "structural_tag": tag_value = constraint[1] # SGLang returns a Pydantic model (LegacyStructuralTagResponseFormat) @@ -1179,6 +1418,8 @@ def __init__( guided_decoding: dict[str, Any] | None = None, response_format_guided_active: bool = False, tool_guided_active: bool = False, + logprobs_enabled: bool = False, + return_tokens_as_token_ids: bool = False, ) -> None: self.tokenizer = tokenizer self.tool_call_parser = tool_call_parser @@ -1205,6 +1446,8 @@ def __init__( response_format_guided_active=response_format_guided_active, tool_guided_active=tool_guided_active, ) + self._logprobs_enabled = logprobs_enabled + self._return_tokens_as_token_ids = return_tokens_as_token_ids self._pending_guided_reasoning_prefix: str | None = ( "" if reasoning_parser is not None and self._guided_json_start_chars @@ -1217,6 +1460,11 @@ def __init__( self._pending_stop_text = "" self._all_token_ids: list[int] = [] + # Logprob records built for chunks that emit no visible delta (e.g. + # a withheld partial UTF-8 character). They are prepended to the + # next emitted choice so probabilities are never dropped with the + # suppressed text (review 20260928 finding 2). + self._withheld_logprobs: list[dict[str, Any]] = [] # Tool call accumulation. SGLang's streaming parser returns # deltas (name in one chunk, argument fragments across subsequent # chunks). However, the base detector processes at most one event @@ -1497,6 +1745,92 @@ def _parse_reasoning_delta( ) return reasoning_text, normal_text, reasoning_tokens + + def _build_logprobs( + self, + token_ids: list[int], + log_probs: list[float] | None, + top_logprobs: list[list[dict[str, Any]]] | None, + ) -> dict[str, Any] | None: + """Build OpenAI-format ``logprobs`` from raw SGLang logprob arrays. + + Returns ``{"content": [...]}`` or ``None`` when no logprobs are + available for this chunk. + """ + if not self._logprobs_enabled or log_probs is None: + return None + + content: list[dict[str, Any]] = [] + for i, lp in enumerate(log_probs): + tid = token_ids[i] if i < len(token_ids) else 0 + if self._return_tokens_as_token_ids: + token_str = f"token_id:{tid}" + else: + try: + token_str = self.tokenizer.decode( + [tid], skip_special_tokens=False + ) + except Exception: + token_str = "" + token_bytes = list(token_str.encode("utf-8")) if token_str else None + + top_list: list[dict[str, Any]] = [] + if top_logprobs and i < len(top_logprobs): + for entry in top_logprobs[i]: + top_tid = entry.get("token_id", 0) + if self._return_tokens_as_token_ids: + top_str = f"token_id:{top_tid}" + else: + top_str = entry.get("token", "") + if not top_str: + try: + top_str = self.tokenizer.decode( + [top_tid], skip_special_tokens=False + ) + except Exception: + top_str = "" + top_bytes = list(top_str.encode("utf-8")) if top_str else None + top_list.append( + { + "token": top_str, + "bytes": top_bytes, + "logprob": entry.get("logprob", 0.0), + } + ) + + content.append( + { + "token": token_str, + "bytes": token_bytes, + "logprob": lp, + "top_logprobs": top_list, + } + ) + + return {"content": content} + + def _withhold_logprobs_payload(self, payload: dict[str, Any] | None) -> None: + """Stash logprob entries when this chunk emits no visible delta. + + A chunk whose text is withheld (unfinished UTF-8 character, stop + string prefix, parser buffering) returns ``None``; without stashing, + its probabilities would be dropped and the stream would end up with + fewer logprob records than completion tokens. + """ + if payload and payload.get("content"): + self._withheld_logprobs.extend(payload["content"]) + + def _release_logprobs_payload( + self, payload: dict[str, Any] | None + ) -> dict[str, Any] | None: + """Prepend withheld entries so emitted records stay in token order.""" + if not self._withheld_logprobs: + return payload + content = self._withheld_logprobs + self._withheld_logprobs = [] + content.extend((payload or {}).get("content") or []) + return {"content": content} + def process_output(self, engine_response: dict[str, Any]) -> dict[str, Any] | None: """Process a single engine response chunk into an OpenAI SSE choice dict. @@ -1513,6 +1847,23 @@ def process_output(self, engine_response: dict[str, Any]) -> dict[str, Any] | No if finished: token_ids = self._strip_trailing_eos_token_ids(list(token_ids)) + # Extract raw logprobs from the engine response. The backend + # (decode_handler / llm_engine) already sliced SGLang's cumulative + # arrays to the new tokens in this chunk, so log_probs and + # top_logprobs align 1:1 with the pre-eos-trim token_ids. + raw_log_probs = engine_response.get("log_probs") + raw_top_logprobs = engine_response.get("top_logprobs") + + # Trim logprobs to match eos-stripped token_ids. + if raw_log_probs and len(raw_log_probs) > len(token_ids): + raw_log_probs = raw_log_probs[: len(token_ids)] + if raw_top_logprobs and len(raw_top_logprobs) > len(token_ids): + raw_top_logprobs = raw_top_logprobs[: len(token_ids)] + + logprobs_payload = self._build_logprobs( + token_ids, raw_log_probs, raw_top_logprobs + ) + # A terminal engine chunk commonly contains no token_ids. Still run # detokenization so a previously withheld suffix can be flushed. delta_text = ( @@ -1530,15 +1881,16 @@ def process_output(self, engine_response: dict[str, Any]) -> dict[str, Any] | No "index": 0, "delta": {"role": "assistant", "content": delta_text}, "finish_reason": finish_reason, - "logprobs": None, + "logprobs": self._release_logprobs_payload(logprobs_payload), } elif finish_reason: return { "index": 0, "delta": {}, "finish_reason": finish_reason, - "logprobs": None, + "logprobs": self._release_logprobs_payload(logprobs_payload), } + self._withhold_logprobs_payload(logprobs_payload) return None # -- Reasoning parsing -- @@ -1856,7 +2208,8 @@ def process_output(self, engine_response: dict[str, Any]) -> dict[str, Any] | No "index": 0, "delta": delta if has_content else {}, "finish_reason": effective_finish, - "logprobs": None, + "logprobs": self._release_logprobs_payload(logprobs_payload), } + self._withhold_logprobs_payload(logprobs_payload) return None diff --git a/dingo/frontend/sglang_processor.py b/dingo/frontend/sglang_processor.py index ecf58a420969..00033b1d15ac 100644 --- a/dingo/frontend/sglang_processor.py +++ b/dingo/frontend/sglang_processor.py @@ -590,6 +590,15 @@ async def _generator_inner( guided_decoding=pre.guided_decoding, response_format_guided_active=pre.response_format_guided_active, tool_guided_active=pre.tool_guided_active, + logprobs_enabled=( + (dynamo_preproc.get("output_options") or {}).get("logprobs") + is not None + ), + return_tokens_as_token_ids=bool( + (dynamo_preproc.get("output_options") or {}).get( + "return_tokens_as_token_ids" + ) + ), ) async for item in self._generate_and_stream( @@ -659,6 +668,17 @@ async def _generator_inner_pool( preproc_result.response_format_guided_active ), tool_guided_active=preproc_result.tool_guided_active, + logprobs_enabled=bool( + (preproc_result.dynamo_preproc.get("output_options") or {}).get( + "logprobs" + ) + is not None + ), + return_tokens_as_token_ids=bool( + (preproc_result.dynamo_preproc.get("output_options") or {}).get( + "return_tokens_as_token_ids" + ) + ), ) async for item in self._generate_and_stream( @@ -697,6 +717,8 @@ async def _generate_and_stream( # TTFT, then switch to the configured interval. pending_token_ids: list[int] = [] pending_usage: dict[str, Any] | None = None + pending_log_probs: list[float] | None = None + pending_top_logprobs: list[list[dict[str, Any]]] | None = None first_chunk = True input_tokens = len(tokens) cumulative_output_tokens = 0 @@ -735,8 +757,16 @@ async def _generate_and_stream( if usage := engine_response.get("completion_usage"): pending_usage = usage - engine_data = engine_response.get("engine_data") pending_token_ids.extend(new_ids) + engine_data = engine_response.get("engine_data") + if chunk_log_probs := engine_response.get("log_probs"): + if pending_log_probs is None: + pending_log_probs = [] + pending_log_probs.extend(chunk_log_probs) + if chunk_top_logprobs := engine_response.get("top_logprobs"): + if pending_top_logprobs is None: + pending_top_logprobs = [] + pending_top_logprobs.extend(chunk_top_logprobs) # Flush on finish or when we've accumulated enough tokens. # First chunk flushes immediately (si=1) to minimize TTFT. @@ -746,6 +776,8 @@ async def _generate_and_stream( mapped_response = { "token_ids": pending_token_ids, "finish_reason": finish_reason, + "log_probs": pending_log_probs, + "top_logprobs": pending_top_logprobs, } if self.debug_perf: @@ -843,13 +875,23 @@ async def _generate_and_stream( if key not in ("usage", "nvext") } payload["choices"] = [{**choice, "finish_reason": None}] - dynamo_out["choices"] = [{**choice, "delta": {}}] + # The logprobs payload rides with the tool-call + # payload only; duplicating it onto the pure finish + # marker would double-count the terminal tokens when + # a client (or the non-stream DeltaAggregator) + # concatenates logprobs across chunks (review + # 20260928 finding 3). + dynamo_out["choices"] = [ + {**choice, "delta": {}, "logprobs": None} + ] yield {**envelope, "data": payload} yield {"_dynamo_annotated": True, "data": dynamo_out} else: yield envelope pending_token_ids = [] + pending_log_probs = None + pending_top_logprobs = None pending_usage = None first_chunk = False except Unknown: diff --git a/dingo/frontend/tests/test_sglang_processor_unit.py b/dingo/frontend/tests/test_sglang_processor_unit.py index c7ab2ae4bfd3..3ef50b39a2aa 100644 --- a/dingo/frontend/tests/test_sglang_processor_unit.py +++ b/dingo/frontend/tests/test_sglang_processor_unit.py @@ -28,6 +28,8 @@ _flatten_message_content, _guided_json_start_chars, _guided_output_requires_reasoning, + _hoist_tool_parameter_defs, + _namespace_tool_parameter_defs, _normalize_assistant_tool_call_arguments, _normalize_prompt_token_ids, _normalize_sglang_parser_name, @@ -1286,6 +1288,467 @@ def test_required_without_parser_name_still_creates_json_array_parser(self): assert isinstance(tcp, JsonArrayParser) +# --------------------------------------------------------------------------- +# _hoist_tool_parameter_defs / _namespace_tool_parameter_defs +# --------------------------------------------------------------------------- + + +def _outer_tool_schema(*tool_params: tuple[str, dict]) -> dict: + """Build an outer tool-call schema shaped like SGLang's named/required + wrapper: each tool appears as an anyOf branch whose properties carry a + ``name`` enum and a ``parameters`` subschema.""" + return { + "type": "array", + "items": { + "anyOf": [ + { + "type": "object", + "properties": { + "name": {"enum": [tool_name]}, + "parameters": params, + }, + } + for tool_name, params in tool_params + ] + }, + } + + +def _branch_params(schema: dict, tool_name: str) -> dict: + for branch in schema["items"]["anyOf"]: + if branch["properties"]["name"]["enum"] == [tool_name]: + return branch["properties"]["parameters"] + raise KeyError(tool_name) + + +class TestHoistToolParameterDefs: # FRONTEND.3 — $defs hoisting collision safety + def test_single_tool_keeps_original_names(self): + schema = _outer_tool_schema( + ( + "get_weather", + { + "type": "object", + "$defs": {"City": {"type": "string"}}, + "properties": {"city": {"$ref": "#/$defs/City"}}, + }, + ) + ) + result = _hoist_tool_parameter_defs(schema) + assert result["$defs"]["City"] == {"type": "string"} + params = _branch_params(result, "get_weather") + assert "$defs" not in params + assert params["properties"]["city"]["$ref"] == "#/$defs/City" + + def test_shared_literal_defs_with_different_dependencies(self): + """Defect repro (review 20260924 f3-1): two tools hold literally + identical ``Payload`` definitions that reference *different* Leaf + schemas. Dict-equality dedup would keep the shared name and let the + later tool's rewritten Payload overwrite the earlier one; every + definition must instead get its own name.""" + a_params = { + "type": "object", + "$defs": { + "Payload": {"$ref": "#/$defs/Leaf"}, + "Leaf": {"type": "string"}, + }, + "properties": {"data": {"$ref": "#/$defs/Payload"}}, + } + b_params = { + "type": "object", + "$defs": { + "Payload": {"$ref": "#/$defs/Leaf"}, + "Leaf": {"type": "integer"}, + }, + "properties": {"data": {"$ref": "#/$defs/Payload"}}, + } + result = _hoist_tool_parameter_defs( + _outer_tool_schema(("a", a_params), ("b", b_params)) + ) + root = result["$defs"] + # a keeps its original names; b is fully renamed. + assert root["Leaf"] == {"type": "string"} + assert root["b__Leaf"] == {"type": "integer"} + assert root["Payload"]["$ref"] == "#/$defs/Leaf" + assert root["b__Payload"]["$ref"] == "#/$defs/b__Leaf" + # References in each tool's parameters follow their own definitions. + assert _branch_params(result, "a")["properties"]["data"]["$ref"] == ( + "#/$defs/Payload" + ) + assert _branch_params(result, "b")["properties"]["data"]["$ref"] == ( + "#/$defs/b__Payload" + ) + + def test_generated_name_does_not_shadow_pending_definition(self): + """Defect repro (review 20260924 f3-2): b defines an integer + ``Payload`` and a boolean ``b__Payload``. Renaming b's ``Payload`` + to ``b__Payload`` must not collide with b's own boolean definition; + whichever is allocated later must move aside.""" + a_params = { + "type": "object", + "$defs": {"Payload": {"type": "string"}}, + "properties": {"data": {"$ref": "#/$defs/Payload"}}, + } + b_params = { + "type": "object", + "$defs": { + "Payload": {"type": "integer"}, + "b__Payload": {"type": "boolean"}, + }, + "properties": { + "num": {"$ref": "#/$defs/Payload"}, + "flag": {"$ref": "#/$defs/b__Payload"}, + }, + } + result = _hoist_tool_parameter_defs( + _outer_tool_schema(("a", a_params), ("b", b_params)) + ) + root = result["$defs"] + assert root["Payload"] == {"type": "string"} + # The integer and the boolean must survive under distinct names. + types = sorted( + d["type"] for name, d in root.items() if name.startswith("b__") + ) + assert types == ["boolean", "integer"] + b = _branch_params(result, "b") + integer_name = b["properties"]["num"]["$ref"].rsplit("/", 1)[-1] + boolean_name = b["properties"]["flag"]["$ref"].rsplit("/", 1)[-1] + assert root[integer_name] == {"type": "integer"} + assert root[boolean_name] == {"type": "boolean"} + assert integer_name != boolean_name + + def test_input_schema_is_not_mutated(self): + params = { + "type": "object", + "$defs": {"City": {"type": "string"}}, + "properties": {"city": {"$ref": "#/$defs/City"}}, + } + schema = _outer_tool_schema(("get_weather", params)) + _hoist_tool_parameter_defs(schema) + assert "$defs" in params # original parameters untouched + + def test_const_literal_with_ref_is_not_rewritten(self): + """Review 20260924 v2: a {"$ref": ...} object inside const/enum is + instance data the model must output, not a schema reference. When + a rename happens (here: the root already holds a builder-merged + City, so the nested copy is renamed), literal values must stay + byte-identical while schema-position refs are rewritten.""" + params = { + "type": "object", + "$defs": {"City": {"type": "string"}}, + "properties": { + "city": {"$ref": "#/$defs/City"}, + "document": {"const": {"$ref": "#/$defs/City"}}, + }, + "required": ["city", "document"], + } + schema = _outer_tool_schema(("submit", params)) + schema["$defs"] = {"City": {"type": "string"}} # builder-merged root + result = _hoist_tool_parameter_defs(schema) + params_out = _branch_params(result, "submit") + # Schema-position ref follows the renamed definition... + assert params_out["properties"]["city"]["$ref"] == ( + "#/$defs/submit__City" + ) + assert result["$defs"]["submit__City"] == {"type": "string"} + # ...but the const literal is instance data and must not change. + assert params_out["properties"]["document"]["const"] == { + "$ref": "#/$defs/City" + } + + def test_enum_literals_preserved_but_anyof_refs_rewritten(self): + """Guard both directions: enum object literals are instance data + (skip), while anyOf arrays hold real subschemas (must rewrite).""" + params = { + "type": "object", + "$defs": {"City": {"type": "string"}}, + "properties": { + "kind": { + "enum": [{"$ref": "#/$defs/City"}, "plain"], + }, + "city": {"anyOf": [{"$ref": "#/$defs/City"}, {"type": "null"}]}, + }, + } + schema = _outer_tool_schema(("submit", params)) + schema["$defs"] = {"City": {"type": "string"}} + result = _hoist_tool_parameter_defs(schema) + params_out = _branch_params(result, "submit") + assert params_out["properties"]["kind"]["enum"] == [ + {"$ref": "#/$defs/City"}, + "plain", + ] + assert params_out["properties"]["city"]["anyOf"][0]["$ref"] == ( + "#/$defs/submit__City" + ) + + def test_pointer_suffix_preserved_on_rename(self): + """A $ref below the definition (#/$defs/City/properties/name) must + keep its JSON-pointer suffix when the leading definition token is + renamed; only the definition name is remapped.""" + params = { + "type": "object", + "$defs": { + "City": { + "type": "object", + "properties": {"name": {"type": "string"}}, + } + }, + "properties": { + "city_name": {"$ref": "#/$defs/City/properties/name"}, + }, + } + schema = _outer_tool_schema(("submit", params)) + # Root already holds a builder-merged City → nested copy is renamed. + schema["$defs"] = { + "City": {"type": "object", "properties": {"name": {"type": "string"}}} + } + result = _hoist_tool_parameter_defs(schema) + params_out = _branch_params(result, "submit") + assert params_out["properties"]["city_name"]["$ref"] == ( + "#/$defs/submit__City/properties/name" + ) + # The renamed definition itself is intact for the suffix to resolve. + assert result["$defs"]["submit__City"]["properties"]["name"] == { + "type": "string" + } + + +class TestNamespaceToolParameterDefs: # FRONTEND.3 — pre-builder $defs isolation + @staticmethod + def _make_tools(*specs: tuple[str, dict]): + return convert_tools( + [ + { + "type": "function", + "function": { + "name": name, + "description": name, + "parameters": params, + }, + } + for name, params in specs + ] + ) + + def test_single_tool_with_defs_passes_through(self): + tools = self._make_tools( + ( + "get_weather", + { + "type": "object", + "$defs": {"City": {"type": "string"}}, + "properties": {"city": {"$ref": "#/$defs/City"}}, + }, + ) + ) + result = _namespace_tool_parameter_defs(tools) + assert result is tools # untouched: no cross-tool conflict possible + + def test_conflicting_names_are_isolated(self): + tools = self._make_tools( + ( + "a", + { + "type": "object", + "$defs": {"Payload": {"type": "string"}}, + "properties": {"data": {"$ref": "#/$defs/Payload"}}, + }, + ), + ( + "b", + { + "type": "object", + "$defs": {"Payload": {"type": "integer"}}, + "properties": {"data": {"$ref": "#/$defs/Payload"}}, + }, + ), + ) + result = _namespace_tool_parameter_defs(tools) + a, b = result + assert set(a.function.parameters["$defs"]) == {"a__Payload"} + assert set(b.function.parameters["$defs"]) == {"b__Payload"} + assert a.function.parameters["properties"]["data"]["$ref"] == ( + "#/$defs/a__Payload" + ) + assert b.function.parameters["properties"]["data"]["$ref"] == ( + "#/$defs/b__Payload" + ) + # Input tools are not mutated (the copy is deep). + assert set(tools[0].function.parameters["$defs"]) == {"Payload"} + assert tools[0].function.parameters["properties"]["data"]["$ref"] == ( + "#/$defs/Payload" + ) + + def test_generated_names_cannot_collide(self): + """tool ``a`` defining ``b__X`` and tool ``a__b`` defining ``X`` + both map to ``a__b__X``; the later one must get a suffix.""" + tools = self._make_tools( + ( + "a", + { + "type": "object", + "$defs": {"b__X": {"type": "string"}}, + "properties": {"v": {"$ref": "#/$defs/b__X"}}, + }, + ), + ( + "a__b", + { + "type": "object", + "$defs": {"X": {"type": "integer"}}, + "properties": {"v": {"$ref": "#/$defs/X"}}, + }, + ), + ) + result = _namespace_tool_parameter_defs(tools) + names_a = set(result[0].function.parameters["$defs"]) + names_b = set(result[1].function.parameters["$defs"]) + assert names_a != names_b + assert not (names_a & names_b) + # References follow each tool's own final name. + for tool in result: + params = tool.function.parameters + final = next(iter(params["$defs"])) + assert params["properties"]["v"]["$ref"] == f"#/$defs/{final}" + + def test_default_and_examples_literals_are_preserved(self): + """Review 20260924 v2: the namespace stage shares the same rewrite + helper, so literal {"$ref": ...} objects in default/examples must + also survive namespacing while schema refs are renamed.""" + tools = self._make_tools( + ( + "a", + { + "type": "object", + "$defs": {"Payload": {"type": "string"}}, + "properties": { + "data": {"$ref": "#/$defs/Payload"}, + "meta": { + "type": "string", + "default": {"$ref": "#/$defs/Payload"}, + "examples": [{"$ref": "#/$defs/Payload"}], + }, + }, + }, + ), + ( + "b", + { + "type": "object", + "$defs": {"Payload": {"type": "integer"}}, + "properties": {"data": {"$ref": "#/$defs/Payload"}}, + }, + ), + ) + result = _namespace_tool_parameter_defs(tools) + meta = result[0].function.parameters["properties"]["meta"] + assert meta["default"] == {"$ref": "#/$defs/Payload"} + assert meta["examples"] == [{"$ref": "#/$defs/Payload"}] + assert result[0].function.parameters["properties"]["data"]["$ref"] == ( + "#/$defs/a__Payload" + ) + + def test_property_named_like_literal_keyword_is_rewritten(self): + """Review 20260928 f1: inside a ``properties`` map the keys are + user-defined names, so a property literally named + const/enum/default/examples still holds a subschema whose ``$ref`` + must be rewritten alongside every other reference.""" + for keyword in ("const", "enum", "default", "examples"): + tools = self._make_tools( + ( + "a", + { + "type": "object", + "$defs": {"Payload": {"type": "string"}}, + "properties": {keyword: {"$ref": "#/$defs/Payload"}}, + }, + ), + ( + "b", + { + "type": "object", + "$defs": {"Payload": {"type": "string"}}, + "properties": {"data": {"$ref": "#/$defs/Payload"}}, + }, + ), + ) + result = _namespace_tool_parameter_defs(tools) + assert set(result[0].function.parameters["$defs"]) == {"a__Payload"} + assert result[0].function.parameters["properties"][keyword][ + "$ref" + ] == "#/$defs/a__Payload", keyword + + def test_keyword_named_property_resolves_through_real_builder(self): + """Review 20260928 f1 end-to-end: identical-content ``$defs`` keep + SGLang's builder from rejecting early, and the final guided schema + must not retain a dangling pointer to the pre-namespace name.""" + tools = self._make_tools( + ( + "a", + { + "type": "object", + "$defs": {"Payload": {"type": "string"}}, + "properties": {"default": {"$ref": "#/$defs/Payload"}}, + "required": ["default"], + }, + ), + ( + "b", + { + "type": "object", + "$defs": {"Payload": {"type": "string"}}, + "properties": {"data": {"$ref": "#/$defs/Payload"}}, + }, + ), + ) + guided = build_tool_call_guided_decoding( + {"tool_choice": "required"}, + tool_call_parser_name=None, + sglang_tools=tools, + ) + assert isinstance(guided, dict) and "json" in guided + schema_text = json.dumps(guided["json"]) + assert "a__Payload" in schema_text + assert "b__Payload" in schema_text + assert '"$ref": "#/$defs/Payload"' not in schema_text + + def test_required_multi_tool_conflicting_defs_real_builder(self): + """Integration through SGLang's real builder (review 20260924 f2): + two tools with same-named, different-content ``$defs`` and + tool_choice='required' previously raised ``ValueError: Tool + definition 'Payload' has multiple schemas`` before any post- + processing could run.""" + tools = self._make_tools( + ( + "a", + { + "type": "object", + "$defs": {"Payload": {"type": "string"}}, + "properties": {"data": {"$ref": "#/$defs/Payload"}}, + }, + ), + ( + "b", + { + "type": "object", + "$defs": {"Payload": {"type": "integer"}}, + "properties": {"data": {"$ref": "#/$defs/Payload"}}, + }, + ), + ) + guided = build_tool_call_guided_decoding( + {"tool_choice": "required"}, + tool_call_parser_name=None, + sglang_tools=tools, + ) + assert isinstance(guided, dict) and "json" in guided + schema_text = json.dumps(guided["json"]) + # Both tools' definitions survive under isolated names, and no + # reference still points at the un-namespaced original. + assert "a__Payload" in schema_text + assert "b__Payload" in schema_text + assert '"$ref": "#/$defs/Payload"' not in schema_text + + # --------------------------------------------------------------------------- # _parse_json_array_buffer # --------------------------------------------------------------------------- @@ -2683,6 +3146,133 @@ def test_fast_path_content_output(self, tokenizer): assert choice["logprobs"] is None +class TestLogprobsStreaming: # FRONTEND.6 — logprobs passthrough integrity + """Logprob records must survive chunks that emit no visible text, and + must not be duplicated when the terminal choice is split in two.""" + + class _ByteTokenizer: + """One token id per raw byte; an incomplete trailing UTF-8 + sequence surfaces as a single held-back U+FFFD, like HF byte-level + decoders feeding the incremental detokenizer.""" + + def decode( + self, token_ids: list[int], *, skip_special_tokens: bool + ) -> str: + text = bytes(token_ids).decode("utf-8", errors="replace") + stripped = text.rstrip("�") + if len(stripped) != len(text): + return stripped + "�" + return text + + def test_logprobs_survive_withheld_text(self): + """Review 20260928 f2: the bytes of one multi-byte character + arriving as separate chunks used to strand every logprob record + except the last byte's.""" + post = SglangStreamingPostProcessor( + tokenizer=self._ByteTokenizer(), + tool_call_parser=None, + reasoning_parser=None, + logprobs_enabled=True, + ) + zhong = list("中".encode("utf-8")) + assert len(zhong) == 3 # the scenario is one character, three bytes + + text = "" + collected: list[float] = [] + for tid, lp in zip(zhong, (-0.1, -0.2, -0.3)): + choice = post.process_output( + {"token_ids": [tid], "log_probs": [lp], "finish_reason": None} + ) + if choice is None: + continue + text += choice["delta"].get("content", "") + collected.extend(e["logprob"] for e in choice["logprobs"]["content"]) + + assert text == "中" + assert collected == [-0.1, -0.2, -0.3] + + # The finish chunk must not re-send already released records. + final = post.process_output({"token_ids": [], "finish_reason": "stop"}) + assert final is not None + if final["logprobs"]: + collected.extend(e["logprob"] for e in final["logprobs"]["content"]) + assert collected == [-0.1, -0.2, -0.3] + + def test_terminal_tool_chunk_sends_logprobs_once(self, tokenizer): + """Review 20260928 f3: the finish/tool-call split attaches + logprobs to the payload choice only, so a client (or the + non-stream DeltaAggregator) concatenating records across chunks + counts the terminal tokens exactly once.""" + + class _TerminalToolCallPost: + """Post-processor double: every chunk looks like the terminal + tool-call flush carrying one logprob record.""" + + def process_output(self, mapped): + return { + "index": 0, + "delta": { + "role": "assistant", + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": {"name": "f", "arguments": "{}"}, + } + ], + }, + "finish_reason": "tool_calls", + "logprobs": { + "content": [ + { + "token": "}", + "bytes": [125], + "logprob": -0.25, + "top_logprobs": [], + } + ] + }, + } + + processor = SglangProcessor( + tokenizer=tokenizer, + routed_engine=FakeRoutedEngine( + items=[{"token_ids": [1], "finish_reason": "stop"}] + ), + tool_call_parser_name=None, + reasoning_parser_name=None, + eos_token_ids=None, + ) + post = _TerminalToolCallPost() + + async def collect(): + return [ + item + async for item in processor._generate_and_stream( + "req-logprobs", {"model": "test-model"}, {}, [], post + ) + ] + + envelopes = asyncio.run(collect()) + data = [e["data"] for e in envelopes if "data" in e] + assert len(data) == 2 + + payload_choice = data[0]["choices"][0] + finish_choice = data[1]["choices"][0] + assert payload_choice["delta"].get("tool_calls") + assert payload_choice["finish_reason"] is None + assert finish_choice["finish_reason"] == "tool_calls" + assert finish_choice["delta"] == {} + + collected = [] + for chunk in data: + lp = chunk["choices"][0].get("logprobs") + if lp: + collected.extend(e["logprob"] for e in lp["content"]) + assert collected == [-0.25] + + # --------------------------------------------------------------------------- # SglangStreamingPostProcessor: reasoning parsing # --------------------------------------------------------------------------- diff --git a/dingo/sglang/CLAUDE.md b/dingo/sglang/CLAUDE.md index 3bb06b9dcb23..177cd2118418 100644 --- a/dingo/sglang/CLAUDE.md +++ b/dingo/sglang/CLAUDE.md @@ -249,18 +249,21 @@ absolute sequence position where logprob computation starts: `-1` (default) = ou only (`len(prompt) - 1`), `0` = from prompt start. We set it to 0 when `prompt_logprobs` is requested. -**Top-logprobs gate**: `logprobs >= 1` (or `prompt_logprobs >= 1`) raises `ValueError` -by default. SGLang's tokenizer manager detokenizes top-k tokens per-position serially, -causing severe latency degradation (O(N) per generated token). Callers must use -`logprobs=0` for chosen-token-only logprobs. Set `DYN_SGL_ALLOW_TOP_LOGPROBS=1` to -override once upstream batches `detokenize_top_logprobs_tokens`. +**Top-logprobs**: `logprobs >= 1` (or `prompt_logprobs >= 1`) is enabled by default. +SGLang versions without batched top-token detokenization may incur extra latency +for long outputs. Set `DYN_SGL_ALLOW_TOP_LOGPROBS=0` to disable top-k logprobs +on deployments where that cost is a concern. **Streaming behavior** (`_extract_logprobs`): -Dynamo forces `stream_output=True` (args.py:374), making `output_ids` disjoint per chunk. -However, SGLang's `meta_info["output_token_logprobs"]` and `meta_info["output_top_logprobs"]` -are always **cumulative** — they grow with each chunk. The handler tracks -`num_output_logprobs_so_far` to slice out only new entries per chunk. +Dynamo forces `stream_output=True` (args.py:374) and +`incremental_streaming_output=True` (args.py:512). In incremental streaming +mode, SGLang's `meta_info["output_token_logprobs"]` and `meta_info["output_top_logprobs"]` +arrive already sliced to the chunk's new tokens (the upstream tokenizer +manager splits them by output offset), so the handler passes +`incremental=True` to `extract_from_sglang_meta()` and forwards them +directly. Only SGLang's default non-incremental mode sends cumulative +arrays that must be sliced by `num_output_logprobs_so_far`. SGLang logprob format: `(logprob, token_id, text_or_None)` tuples. Dynamo output format: `log_probs` = list of floats, `top_logprobs` = list of lists of @@ -308,9 +311,11 @@ text-to-video-diffusion.sh # 1-2 GPUs - Text-to-video (Wan2.1) - **output_modalities default**: Global default is `["text"]`. Image/video diffusion workers must override to `["image"]`/`["video"]` or the Rust registration path tries to load `config.json` (which doesn't exist for diffusers models). -- **Cumulative logprobs in streaming**: SGLang's `output_token_logprobs`/`output_top_logprobs` - in `meta_info` are cumulative even though `output_ids` are disjoint (stream_output=True). - Always slice with an offset, don't assume per-chunk logprobs. +- **Per-chunk logprobs in incremental streaming**: with Dynamo's forced + `incremental_streaming_output=True`, `output_token_logprobs`/`output_top_logprobs` + in `meta_info` are already disjoint per chunk — forward them directly. + Treating them as cumulative and slicing with `num_output_logprobs_so_far` + returns nothing for every chunk after the first. - **Zombie GPU processes**: `sgl_diffusion::scheduler` spawns a child process that survives parent kill. Always check `nvidia-smi` after teardown. - **Session radix cache**: SGLang 0.5.14+ provides session-aware radix ownership diff --git a/dingo/sglang/llm_engine.py b/dingo/sglang/llm_engine.py index b4f0758b04ab..3aa2f5edd6ed 100644 --- a/dingo/sglang/llm_engine.py +++ b/dingo/sglang/llm_engine.py @@ -481,6 +481,7 @@ async def generate( meta_info, num_logprobs_per_choice.get(output_idx, 0), return_tokens_as_token_ids=return_tokens_as_token_ids, + incremental=True, ) num_logprobs_per_choice[output_idx] = next_total if log_probs is not None: diff --git a/dingo/sglang/request_handlers/llm/decode_handler.py b/dingo/sglang/request_handlers/llm/decode_handler.py index 504268f19945..2e80909d8d6d 100644 --- a/dingo/sglang/request_handlers/llm/decode_handler.py +++ b/dingo/sglang/request_handlers/llm/decode_handler.py @@ -456,6 +456,7 @@ def _extract_logprobs( meta_info, num_output_logprobs_so_far, return_tokens_as_token_ids=return_tokens_as_token_ids, + incremental=True, ) async def generate( diff --git a/dingo/sglang/tests/test_sglang_decode_handler.py b/dingo/sglang/tests/test_sglang_decode_handler.py index 41c6528867b4..4b321fcd8478 100644 --- a/dingo/sglang/tests/test_sglang_decode_handler.py +++ b/dingo/sglang/tests/test_sglang_decode_handler.py @@ -436,21 +436,20 @@ def test_build_logprob_kwargs_allows_chosen_token_logprobs(monkeypatch): assert kwargs == {"return_logprob": True, "top_logprobs_num": 0} -def test_build_logprob_kwargs_rejects_top_logprobs_by_default(monkeypatch): +def test_build_logprob_kwargs_allows_top_logprobs_by_default(monkeypatch): monkeypatch.delenv("DYN_SGL_ALLOW_TOP_LOGPROBS", raising=False) - with pytest.raises(ValueError, match="does not currently support logprobs >= 1"): - DecodeWorkerHandler._build_logprob_kwargs({"output_options": {"logprobs": 1}}) - - -def test_build_logprob_kwargs_allows_top_logprobs_with_escape_hatch(monkeypatch): - monkeypatch.setenv("DYN_SGL_ALLOW_TOP_LOGPROBS", "1") - kwargs = DecodeWorkerHandler._build_logprob_kwargs( - {"output_options": {"logprobs": 2}} + {"output_options": {"logprobs": 1}} ) + assert kwargs == {"return_logprob": True, "top_logprobs_num": 1} - assert kwargs == {"return_logprob": True, "top_logprobs_num": 2} + +def test_build_logprob_kwargs_rejects_top_logprobs_when_disabled(monkeypatch): + monkeypatch.setenv("DYN_SGL_ALLOW_TOP_LOGPROBS", "0") + + with pytest.raises(ValueError, match="disabled by DYN_SGL_ALLOW_TOP_LOGPROBS=0"): + DecodeWorkerHandler._build_logprob_kwargs({"output_options": {"logprobs": 1}}) def test_extract_logprobs_formats_top_tokens_as_token_ids(): diff --git a/docs/tool-calling/troubleshooting.md b/docs/tool-calling/troubleshooting.md index ffc3e22d8c18..ffc444ca25b1 100644 --- a/docs/tool-calling/troubleshooting.md +++ b/docs/tool-calling/troubleshooting.md @@ -25,11 +25,10 @@ diagnostic info. > the assistant-content channel; the reasoning channel is not surfaced in > `logprobs.content`. > -> If the worker is the SGLang backend, `logprobs: true` is rejected by -> default because SGLang's tokenizer manager detokenizes top-k tokens -> serially, causing latency degradation. Launch the worker with -> `DYN_SGL_ALLOW_TOP_LOGPROBS=1` set in the environment to opt in for the -> duration of the repro request, then unset it afterward. Tracked at +> SGLang top-k logprobs are enabled by default. SGLang versions without +> batched top-token detokenization may incur extra latency for long outputs. +> Set `DYN_SGL_ALLOW_TOP_LOGPROBS=0` on the worker to disable them if needed. +> The upstream optimization proposal is described at > [sgl-project/sglang#24447](https://github.com/sgl-project/sglang/pull/24447). ## The request