Skip to content
Merged
2 changes: 1 addition & 1 deletion components/src/dynamo/sglang/init_embedding.py
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,7 @@ async def init_embedding(
server_args,
dynamo_args,
input_type=ModelInput.Text,
output_type=ModelType.Embedding,
output_type=ModelType.Embedding | ModelType.Rerank,
Comment thread
xianlubird marked this conversation as resolved.
Resolved
readiness_gate=ready_event,
worker_type=WorkerType.Aggregated,
needs=[],
Expand Down
16 changes: 13 additions & 3 deletions components/src/dynamo/sglang/protocol.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,12 +57,22 @@ class EmbeddingRequest(BaseModel):
model: str
input: EmbeddingInput
user: Optional[str] = None
dimensions: Optional[
int
] = None # only supported in text-embedding-3 and later models from OpenAI
dimensions: Optional[int] = (
None # only supported in text-embedding-3 and later models from OpenAI
)
encoding_format: Literal["float", "base64"] = "float"


class RerankRequest(BaseModel):
"""SGLang-compatible text-only cross-encoder rerank request."""

model: str
query: str
documents: List[str]
top_n: Optional[int] = None
return_documents: bool = True


class DisaggPreprocessedRequest(BaseModel):
request: Union[PreprocessedRequest, ChatCompletionRequest]
sampling_params: dict
Expand Down
2 changes: 1 addition & 1 deletion components/src/dynamo/sglang/register.py
Original file line number Diff line number Diff line change
Expand Up @@ -151,7 +151,7 @@ async def _register_model_with_runtime_config(
)
input_type = ModelInput.Text
# Only override output_type for chat models, not for embeddings
if output_type != ModelType.Embedding:
if not output_type.supports_embedding():
output_type = ModelType.Chat

if runtime_config is not None and _supports_engine_generate(
Expand Down
2 changes: 2 additions & 0 deletions components/src/dynamo/sglang/request_handlers/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
"PrefillWorkerHandler": ".llm",
# Embedding handlers
"EmbeddingWorkerHandler": ".embedding",
"RerankWorkerHandler": ".rerank",
# Image diffusion handlers
"ImageDiffusionWorkerHandler": ".image_diffusion",
# Video generation handlers
Expand Down Expand Up @@ -45,6 +46,7 @@ def __getattr__(name: str) -> Any:
"PrefillWorkerHandler",
# Embedding handlers
"EmbeddingWorkerHandler",
"RerankWorkerHandler",
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
# Image diffusion handlers
"ImageDiffusionWorkerHandler",
# Video generation handlers
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
observe_embedding_input_tokens,
)
from dynamo.sglang.request_handlers.handler_base import BaseWorkerHandler
from dynamo.sglang.request_handlers.rerank import RerankWorkerHandler


def _encode_floats_to_base64(floats: List[float]) -> str:
Expand All @@ -44,6 +45,9 @@ def __init__(
shutdown_event: Optional[asyncio.Event] = None,
):
super().__init__(engine, config, publisher, None, shutdown_event)
self.rerank_handler = RerankWorkerHandler(
engine, enable_trace=self.enable_trace
)
logging.info("Embedding worker handler initialized")

def cleanup(self) -> None:
Expand Down Expand Up @@ -97,6 +101,11 @@ async def generate(
request: Embedding request dictionary.
context: Context object for cancellation handling.
"""
if "query" in request or "documents" in request:
async for response in self.rerank_handler.generate(request, context):
yield response
return

embedding_input = request.get("input")
if isinstance(embedding_input, str):
input_type = "str"
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

from dynamo.sglang.request_handlers.rerank.rerank_handler import RerankWorkerHandler

__all__ = ["RerankWorkerHandler"]
Original file line number Diff line number Diff line change
@@ -0,0 +1,124 @@
# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

import heapq
from collections.abc import AsyncGenerator
from typing import Any

import sglang as sgl
from sglang.srt.managers.io_struct import EmbeddingReqInput

from dynamo._core import Context
from dynamo.sglang.protocol import RerankRequest


def _looks_like_decoder_reranker(tokenizer_manager: Any) -> bool:
"""Detect decoder-only Qwen3 rerankers that need SGLang's generation path."""
tokenizer = getattr(tokenizer_manager, "tokenizer", None)
chat_template = getattr(tokenizer, "chat_template", "")
template = chat_template.lower() if isinstance(chat_template, str) else ""
model_config = getattr(tokenizer_manager, "model_config", None)
model_path = str(getattr(model_config, "model_path", "")).lower()
has_yes_no_template = (
"answer can only be" in template and "yes" in template and "no" in template
)
return (
has_yes_no_template
or "qwen3-reranker" in model_path
or "qwen3-vl-reranker" in model_path
)


class RerankWorkerHandler:
"""Runs stable text-only cross-encoder reranking through SGLang pooling."""

def __init__(self, engine: sgl.Engine, *, enable_trace: bool) -> None:
self.engine = engine
self.enable_trace = enable_trace

@staticmethod
def _validate(request: RerankRequest) -> None:
if not request.query.strip():
raise ValueError("Query cannot be empty or whitespace only")
if not request.documents:
raise ValueError("Documents cannot be empty")
if any(not document.strip() for document in request.documents):
raise ValueError("Each document cannot be empty or whitespace only")
if request.top_n is not None and request.top_n < 1:
raise ValueError("Parameter 'top_n' must be larger than 0")

@staticmethod
def _score(item: Any, index: int) -> float:
if not isinstance(item, dict) or "embedding" not in item:
raise ValueError(f"Missing embedding score for rerank at index {index}")
score = item["embedding"]
if isinstance(score, list):
if not score or not isinstance(score[0], (int, float)):
raise ValueError(
f"Invalid embedding score for rerank at index {index}: {score!r}"
)
score = score[0]
if not isinstance(score, (int, float)):
raise ValueError(
f"Invalid embedding score for rerank at index {index}: {score!r}"
)
return float(score)

async def generate(
self, request: dict[str, Any], context: Context
) -> AsyncGenerator[list[dict[str, Any]], None]:
rerank_request = RerankRequest(**request)
self._validate(rerank_request)

tokenizer_manager = self.engine.tokenizer_manager
if _looks_like_decoder_reranker(tokenizer_manager):
raise ValueError(
"Dynamo's SGLang /v1/rerank integration currently supports "
"cross-encoder rerankers only; decoder-only Qwen3 and Qwen3-VL "
"rerankers are not supported"
)

pairs = [
[rerank_request.query, document] for document in rerank_request.documents
]
trace_id = context.trace_id
request_ids = (
[f"{trace_id}-{index}" for index in range(len(pairs))]
if trace_id is not None
else None
)
internal_request = EmbeddingReqInput(
text=pairs,
is_cross_encoder_request=True,
external_trace_header=(
context.trace_headers() if self.enable_trace else None
),
rid=request_ids,
)
responses = tokenizer_manager.generate_request(internal_request, None)
result = await anext(responses)
if not isinstance(result, list):
result = [result]
if len(result) != len(rerank_request.documents):
raise ValueError(
"SGLang returned an unexpected number of rerank scores: "
f"expected {len(rerank_request.documents)}, got {len(result)}"
)

ranked = []
for index, item in enumerate(result):
response = {"score": self._score(item, index), "index": index}
if rerank_request.return_documents:
response["document"] = rerank_request.documents[index]
meta_info = item.get("meta_info") if isinstance(item, dict) else None
if meta_info is not None:
response["meta_info"] = meta_info
ranked.append(response)

if rerank_request.top_n is not None:
ranked = heapq.nlargest(
rerank_request.top_n, ranked, key=lambda item: item["score"]
)
else:
ranked.sort(key=lambda item: item["score"], reverse=True)
yield ranked
175 changes: 175 additions & 0 deletions components/src/dynamo/sglang/tests/test_sglang_rerank_handler.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,175 @@
# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Unit tests for SGLang cross-encoder reranking."""

from types import SimpleNamespace
from typing import Any

import pytest

pytest.importorskip(
"sglang.srt.managers.io_struct", reason="sglang not installed in this container"
)

from dynamo.sglang.request_handlers.embedding import ( # noqa: E402
EmbeddingWorkerHandler,
)
from dynamo.sglang.request_handlers.rerank import RerankWorkerHandler # noqa: E402

pytestmark = [
pytest.mark.unit,
pytest.mark.sglang,
pytest.mark.core,
pytest.mark.gpu_0,
pytest.mark.profiled_vram_gib(0),
pytest.mark.pre_merge,
]


class _TokenizerManager:
def __init__(
self,
scores: list[float],
*,
chat_template: str = "",
model_path: str = "BAAI/bge-reranker-v2-m3",
) -> None:
self.scores = scores
self.requests: list[tuple[Any, Any]] = []
self.tokenizer = SimpleNamespace(chat_template=chat_template)
self.model_config = SimpleNamespace(model_path=model_path)

async def generate_request(self, request: Any, raw_request: Any):
request.normalize_batch_and_arguments()
self.requests.append((request, raw_request))
yield [
{
"embedding": [score],
"meta_info": {"prompt_tokens": index + 1},
}
for index, score in enumerate(self.scores)
]


class _Engine:
def __init__(self, tokenizer_manager: _TokenizerManager) -> None:
self.tokenizer_manager = tokenizer_manager


class _Context:
trace_id = "rerank-trace"

def trace_headers(self) -> dict[str, str]:
return {"traceparent": "00-test"}


def _handler(
scores: list[float],
*,
chat_template: str = "",
model_path: str = "BAAI/bge-reranker-v2-m3",
) -> tuple[RerankWorkerHandler, _TokenizerManager]:
manager = _TokenizerManager(
scores, chat_template=chat_template, model_path=model_path
)
return RerankWorkerHandler(_Engine(manager), enable_trace=True), manager


@pytest.mark.asyncio
async def test_builds_pairs_sorts_scores_and_applies_top_n():
handler, manager = _handler([0.2, 0.9, 0.5])
outputs = [
output
async for output in handler.generate(
{
"model": "reranker",
"query": "query",
"documents": ["zero", "one", "two"],
"top_n": 2,
"return_documents": True,
},
_Context(),
)
]

assert [item["index"] for item in outputs[0]] == [1, 2]
assert [item["document"] for item in outputs[0]] == ["one", "two"]
[(request, raw_request)] = manager.requests
assert raw_request is None
assert request.text == [["query", "zero"], ["query", "one"], ["query", "two"]]
assert request.is_cross_encoder_request is True
assert request.rid == ["rerank-trace-0", "rerank-trace-1", "rerank-trace-2"]
assert request.external_trace_header == {"traceparent": "00-test"}


@pytest.mark.asyncio
async def test_omits_documents_when_not_requested():
handler, _ = _handler([0.1, 0.8])
[output] = [
output
async for output in handler.generate(
{
"model": "reranker",
"query": "query",
"documents": ["a", "b"],
"return_documents": False,
},
_Context(),
)
]
assert [item["index"] for item in output] == [1, 0]
assert all("document" not in item for item in output)


@pytest.mark.asyncio
@pytest.mark.parametrize(
"request",
[
{"model": "m", "query": " ", "documents": ["doc"]},
{"model": "m", "query": "query", "documents": []},
{"model": "m", "query": "query", "documents": [""]},
{"model": "m", "query": "query", "documents": ["doc"], "top_n": 0},
],
)
async def test_rejects_invalid_requests_before_inference(request):
handler, manager = _handler([0.5])
with pytest.raises(ValueError):
_ = [output async for output in handler.generate(request, _Context())]
assert manager.requests == []


@pytest.mark.asyncio
async def test_rejects_decoder_only_qwen3_reranker():
handler, manager = _handler(
[0.5],
chat_template='The answer can only be "yes" or "no"',
model_path="Qwen/Qwen3-Reranker-0.6B",
)
with pytest.raises(ValueError, match="cross-encoder rerankers only"):
_ = [
output
async for output in handler.generate(
{"model": "m", "query": "query", "documents": ["doc"]},
_Context(),
)
]
assert manager.requests == []


@pytest.mark.asyncio
async def test_embedding_endpoint_dispatches_rerank_shape():
class _RerankHandler:
async def generate(self, request, context):
yield [{"score": 0.7, "index": 0}]

handler = EmbeddingWorkerHandler.__new__(EmbeddingWorkerHandler)
handler.rerank_handler = _RerankHandler()
output = [
item
async for item in handler.generate(
{"model": "m", "query": "query", "documents": ["doc"]},
_Context(),
)
]
assert output == [[{"score": 0.7, "index": 0}]]
Loading
Loading