sgl-project--sglang/python/sglang/srt/entrypoints/openai/serving_rerank.py

import heapq
import logging
import math
from typing import Any, Dict, List, Optional, Union

from fastapi import Request
from fastapi.responses import ORJSONResponse

from sglang.srt.entrypoints.openai.protocol import (
    ChatCompletionMessageContentImagePart,
    ChatCompletionMessageContentTextPart,
    ChatCompletionMessageContentVideoPart,
    ErrorResponse,
    RerankContent,
    RerankResponse,
    V1RerankReqInput,
)
from sglang.srt.entrypoints.openai.serving_base import OpenAIServingBase
from sglang.srt.managers.io_struct import EmbeddingReqInput, GenerateReqInput

logger = logging.getLogger(__name__)


def _get_yes_no_token_ids(tokenizer) -> tuple[int, int]:
    """Get token IDs for 'yes' and 'no' from the tokenizer.

    Different model sizes may have different token IDs, so we look them up dynamically.
    """
    # Try to encode 'yes' and 'no' to get their token IDs
    # The tokenizer should return a single token for these common words
    try:
        yes_tokens = tokenizer.encode("yes", add_special_tokens=False)
        no_tokens = tokenizer.encode("no", add_special_tokens=False)

        if len(yes_tokens) == 1 and len(no_tokens) == 1:
            return yes_tokens[0], no_tokens[0]

        # Fallback: try convert_tokens_to_ids
        yes_id = tokenizer.convert_tokens_to_ids("yes")
        no_id = tokenizer.convert_tokens_to_ids("no")
        if yes_id is not None and no_id is not None:
            return yes_id, no_id

    except Exception as e:
        logger.warning(f"Failed to get yes/no token IDs dynamically: {e}")

    # Fallback to known Qwen3 token IDs (may not work for all model sizes)
    logger.warning("Using fallback token IDs for yes/no (9693/2152)")
    return 9693, 2152


def _is_qwen3_reranker_template(chat_template: str) -> bool:
    """Detect if the chat template is for Qwen3 text-only reranker."""
    if not chat_template:
        return False
    t = chat_template.lower()
    return ('answer can only be "yes" or "no"' in t) or (
        "answer can only be" in t and '"yes"' in t and '"no"' in t
    )


def _is_qwen3_vl_reranker_template(chat_template: str) -> bool:
    """Detect if the chat template is for Qwen3-VL multimodal reranker.

    VL reranker templates use `query` and `document` as jinja variables
    and include vision token placeholders for image/video support.
    """
    if not chat_template:
        return False
    t = chat_template.lower()
    # Check for reranker phrase (yes/no judgment)
    has_reranker_phrase = ('answer can only be "yes" or "no"' in t) or (
        "answer can only be" in t and '"yes"' in t and '"no"' in t
    )
    # Check for vision token placeholders (unique to VL templates)
    has_vision_tokens = "<|vision_start|>" in t or "<|image_pad|>" in t
    return has_reranker_phrase and has_vision_tokens


def _is_qwen3_vl_model(model_path: str) -> bool:
    """Check if the model is a Qwen3-VL model based on model path."""
    if not model_path:
        return False
    model_lower = model_path.lower()
    return "qwen3-vl" in model_lower or "qwen3vl" in model_lower


def _detect_rerank_backend(
    *,
    request: V1RerankReqInput,
    chat_template: Optional[str],
    model_path: str,
) -> str:
    """
    Unify rerank routing decisions used by both `_convert_to_internal_request` and
    `_handle_non_streaming_request`.

    Returns:
        "vl_decoder" | "text_decoder" | "cross_encoder"
    """
    is_multimodal = request.is_multimodal()
    is_vl_model = _is_qwen3_vl_model(model_path)
    is_vl_template = _is_qwen3_vl_reranker_template(chat_template)
    is_text_template = _is_qwen3_reranker_template(chat_template)

    # Prefer VL when template/model indicates VL, or request is multimodal with reranker template.
    if is_vl_template or is_vl_model or (is_multimodal and is_text_template):
        return "vl_decoder"
    if is_text_template:
        return "text_decoder"
    return "cross_encoder"


def _qwen3_rerank_score(p_yes: float, p_no: float) -> float:
    denom = p_yes + p_no
    if denom <= 0.0:
        return 0.0
    return p_yes / denom


def _get_jinja_env():
    try:
        import jinja2  # Lazy import: server env should provide this dependency.
        from jinja2.sandbox import ImmutableSandboxedEnvironment
    except ModuleNotFoundError as e:
        raise ValueError(
            "Rendering Qwen3 reranker prompts requires `jinja2`. "
            "Please install it in your runtime environment (e.g., `pip install jinja2`)."
        ) from e
    # Using a sandboxed environment to stop malicious execution during model loading.
    return ImmutableSandboxedEnvironment(
        loader=jinja2.BaseLoader(),
        autoescape=False,
        undefined=jinja2.Undefined,
    )


def _render_jinja_chat_template(
    chat_template: str,
    *,
    query: RerankContent,
    document: RerankContent,
    instruct: Optional[str],
) -> str:
    """Render a loaded Jinja chat template for Qwen3 reranker prompts (text-only)."""
    env = _get_jinja_env()
    template = env.from_string(chat_template)

    # For text-only template, extract text content
    query_text = query if isinstance(query, str) else _extract_text_from_content(query)
    doc_text = (
        document if isinstance(document, str) else _extract_text_from_content(document)
    )

    render_kwargs = {
        "messages": [
            {"role": "user", "content": query_text},
            {"role": "user", "content": doc_text},
        ]
    }
    # Only pass instruct when explicitly provided; template uses `default(...)`
    # which works only when the variable is undefined (not None).
    if instruct:
        render_kwargs["instruct"] = instruct
    return template.render(**render_kwargs)


def _render_vl_jinja_template(
    chat_template: str,
    *,
    query: List[Dict[str, Any]],
    document: List[Dict[str, Any]],
    instruct: Optional[str],
) -> str:
    """Render a loaded Jinja chat template for Qwen3-VL reranker prompts (multimodal).

    The template expects `query` and `document` as lists of content parts,
    where each part has a `type` field (text, image, video) and corresponding data.
    """
    env = _get_jinja_env()
    template = env.from_string(chat_template)

    render_kwargs = {
        "query": query,
        "document": document,
    }
    if instruct:
        render_kwargs["instruct"] = instruct
    return template.render(**render_kwargs)


def _extract_text_from_content(content: RerankContent) -> str:
    """Extract text from multimodal content."""
    if isinstance(content, str):
        return content
    texts = []
    for part in content:
        if isinstance(part, ChatCompletionMessageContentTextPart):
            texts.append(part.text)
        elif isinstance(part, dict) and part.get("type") == "text":
            texts.append(part.get("text", ""))
    return " ".join(texts)


class OpenAIServingRerank(OpenAIServingBase):
    """Handler for /v1/rerank requests"""

    def __init__(self, tokenizer_manager, template_manager=None):
        super().__init__(tokenizer_manager)
        # TemplateManager is optional; rerank uses tokenizer.chat_template today.
        # Keeping this explicit makes the dependency clear and supports future extensions.
        self.template_manager = template_manager

        # Cache yes/no token IDs for Qwen3 reranker scoring
        self._yes_token_id, self._no_token_id = _get_yes_no_token_ids(
            tokenizer_manager.tokenizer
        )

    # NOTE: /v1/rerank is not an official OpenAI endpoint. This module may be moved
    # to another module in the future.

    def _request_id_prefix(self) -> str:
        return "rerank-"

    def _validate_request(self, request: V1RerankReqInput) -> Optional[str]:
        """Validate rerank request format and content"""
        if not request.query:
            return "Query cannot be empty"

        if isinstance(request.query, str):
            if not request.query.strip():
                return "Query cannot be empty or whitespace only"

        if not request.documents:
            return "Documents cannot be empty"

        for doc in request.documents:
            if not doc:
                return "Each document must be a non-empty string"
            if isinstance(doc, str) and not doc.strip():
                return "Each document cannot be empty or whitespace only"

        return None

    def _convert_to_internal_request(
        self,
        request: V1RerankReqInput,
        raw_request: Request = None,
    ) -> tuple[Union[EmbeddingReqInput, V1RerankReqInput], V1RerankReqInput]:
        """
        Convert OpenAI rerank request to internal format.

        - For Qwen3-VL reranker (multimodal decoder-only): keep the request.
        - For Qwen3 reranker (text-only decoder-only): keep the request and score via
          `tokenizer_manager.score_prompts(...)` in the handler.
        - For cross-encoder rerank models: adapt into `EmbeddingReqInput` pairs.
        """
        chat_template = self.tokenizer_manager.tokenizer.chat_template
        model_path = getattr(self.tokenizer_manager.model_config, "model_path", "")
        backend = _detect_rerank_backend(
            request=request,
            chat_template=chat_template if isinstance(chat_template, str) else None,
            model_path=model_path,
        )
        if backend in ("vl_decoder", "text_decoder"):
            return request, request

        # Cross-encoder rerank: Create pairs of [query, document] for each document.
        # Note: Cross-encoder only supports text-only content
        if request.is_multimodal():
            # Extract text for cross-encoder (multimodal not supported)
            query_text = _extract_text_from_content(request.query)
            doc_texts = [_extract_text_from_content(doc) for doc in request.documents]
            pairs = [[query_text, doc] for doc in doc_texts]
        else:
            pairs = [[request.query, doc] for doc in request.documents]

        adapted_request = EmbeddingReqInput(text=pairs, is_cross_encoder_request=True)
        return adapted_request, request

    async def _handle_non_streaming_request(
        self,
        adapted_request: Union[EmbeddingReqInput, V1RerankReqInput],
        request: V1RerankReqInput,
        raw_request: Request,
    ) -> Union[List[RerankResponse], ErrorResponse, ORJSONResponse]:
        """Handle the rerank request"""
        chat_template = getattr(self.tokenizer_manager.tokenizer, "chat_template", None)
        model_path = getattr(self.tokenizer_manager.model_config, "model_path", "")
        rerank_ret = await self._handle_rerank_paths(
            request=request,
            raw_request=raw_request,
            chat_template=chat_template,
            model_path=model_path,
        )
        if rerank_ret is not None:
            return rerank_ret

        # Default cross-encoder rerank path (existing behavior).
        try:
            if not isinstance(adapted_request, EmbeddingReqInput):
                raise ValueError(
                    "Invalid rerank request adaptation. "
                    "If you are serving a decoder-only reranker (e.g., Qwen3-Reranker), "
                    "please provide the corresponding --chat-template and launch without --is-embedding."
                )
            ret = await self.tokenizer_manager.generate_request(
                adapted_request, raw_request
            ).__anext__()
        except ValueError as e:
            return self.create_error_response(str(e))

        if not isinstance(ret, list):
            ret = [ret]

        responses = self._build_rerank_response(ret, request)
        return responses

    async def _handle_rerank_paths(
        self,
        *,
        request: V1RerankReqInput,
        raw_request: Request,
        chat_template: Optional[str],
        model_path: str,
    ) -> Optional[Union[List[RerankResponse], ErrorResponse, ORJSONResponse]]:
        """
        Handle decoder-only rerank paths (VL/text) and return a response if matched.

        Returns None if the request should fall back to cross-encoder rerank.
        """
        backend = _detect_rerank_backend(
            request=request,
            chat_template=chat_template,
            model_path=model_path,
        )

        # Qwen3-VL reranker path (decoder-only scoring with query/document template format)
        if backend == "vl_decoder":
            return await self._handle_vl_reranker_request(
                request, raw_request, chat_template or ""
            )

        # Qwen3 text-only reranker path (decoder-only scoring).
        if backend == "text_decoder":
            return await self._handle_text_reranker_request(
                request=request,
                raw_request=raw_request,
                chat_template=chat_template or "",
            )

        return None

    async def _handle_text_reranker_request(
        self,
        *,
        request: V1RerankReqInput,
        raw_request: Request,
        chat_template: str,
    ) -> Union[List[RerankResponse], ErrorResponse]:
        """Handle text-only decoder reranker request via score_prompts()."""
        # Qwen3 reranker relies on decoder-only logprobs. If the server is launched
        # with --is-embedding, model_config.is_generation is typically False and
        # logprob scoring is not supported.
        if not self.tokenizer_manager.model_config.is_generation:
            return self.create_error_response(
                "Detected Qwen3 reranker chat template, but the server is not in generation mode. "
                "Please relaunch without --is-embedding for Qwen3-Reranker models."
            )

        try:
            prompts = [
                _render_jinja_chat_template(
                    chat_template,
                    query=request.query,
                    document=doc,
                    instruct=getattr(request, "instruct", None),
                )
                for doc in request.documents
            ]

            result = await self.tokenizer_manager.score_prompts(
                prompts,
                label_token_ids=[self._yes_token_id, self._no_token_id],
                apply_softmax=False,
                request=raw_request,
            )
            scores = [_qwen3_rerank_score(s[0], s[1]) for s in result.scores]
        except ValueError as e:
            return self.create_error_response(str(e))
        except Exception as e:
            # Includes template rendering errors from jinja2.
            return self.create_error_response(str(e))

        responses = self._build_rerank_response(scores, request)
        return responses

    async def _handle_vl_reranker_request(
        self,
        request: V1RerankReqInput,
        raw_request: Request,
        _chat_template: str,
    ) -> Union[List[RerankResponse], ErrorResponse]:
        """Handle multimodal VL reranker request using chat completion with logprobs."""
        if not self.tokenizer_manager.model_config.is_generation:
            return self.create_error_response(
                "Detected Qwen3-VL reranker, but the server is not in generation mode. "
                "Please relaunch without --is-embedding for Qwen3-VL-Reranker models."
            )

        try:
            scores = []
            instruct = getattr(request, "instruct", None)

            for doc in request.documents:
                # Build multimodal content lists and render prompt using jinja template
                query_content, doc_content, image_data, video_data = (
                    self._build_vl_reranker_content(
                        query=request.query,
                        document=doc,
                    )
                )

                # Render the chat template directly with query/document variables
                prompt = _render_vl_jinja_template(
                    chat_template=_chat_template,
                    query=query_content,
                    document=doc_content,
                    instruct=instruct,
                )

                # Create generate request with logprobs
                gen_request = GenerateReqInput(
                    text=prompt,
                    image_data=image_data if image_data else None,
                    video_data=video_data if video_data else None,
                    sampling_params={
                        "max_new_tokens": 1,
                        "temperature": 0,
                    },
                    return_logprob=True,
                    top_logprobs_num=50,  # Get enough logprobs to find yes/no tokens
                    logprob_start_len=0,
                )

                # Execute generation request
                ret = await self.tokenizer_manager.generate_request(
                    gen_request, raw_request
                ).__anext__()

                # Extract yes/no probabilities from logprobs
                score = self._extract_score_from_logprobs(ret)
                scores.append(score)

            responses = self._build_rerank_response(scores, request)
            return responses

        except ValueError as e:
            return self.create_error_response(str(e))
        except Exception as e:
            logger.exception("Error handling VL reranker request")
            return self.create_error_response(str(e))

    def _build_vl_reranker_content(
        self,
        query: RerankContent,
        document: RerankContent,
    ) -> tuple[List[Dict[str, Any]], List[Dict[str, Any]], List[str], List[str]]:
        """Build content lists for VL reranker request.

        Returns:
            Tuple of (query_content, document_content, image_data, video_data)
            where query_content and document_content are lists suitable for jinja template.
        """
        image_data = []
        video_data = []

        # Build query content list
        query_content = self._content_to_template_list(query, image_data, video_data)

        # Build document content list
        doc_content = self._content_to_template_list(document, image_data, video_data)

        return query_content, doc_content, image_data, video_data

    def _content_to_template_list(
        self,
        content: RerankContent,
        image_data: List[str],
        video_data: List[str],
    ) -> List[Dict[str, Any]]:
        """Convert RerankContent to a list format suitable for jinja template."""
        result = []

        if isinstance(content, str):
            result.append({"type": "text", "text": content})
            return result

        for part in content:
            if isinstance(part, ChatCompletionMessageContentTextPart):
                result.append({"type": "text", "text": part.text})
            elif isinstance(part, ChatCompletionMessageContentImagePart):
                if part.image_url:
                    image_data.append(part.image_url.url)
                    result.append({"type": "image"})
            elif isinstance(part, ChatCompletionMessageContentVideoPart):
                if part.video_url:
                    video_data.append(part.video_url.url)
                    result.append({"type": "video"})
            elif isinstance(part, dict):
                part_type = part.get("type")
                if part_type == "text":
                    result.append({"type": "text", "text": part.get("text", "")})
                elif part_type == "image_url":
                    image_url = part.get("image_url", {})
                    if isinstance(image_url, dict):
                        url = image_url.get("url")
                    else:
                        url = image_url
                    if url:
                        image_data.append(url)
                        result.append({"type": "image"})
                elif part_type == "video_url":
                    video_url = part.get("video_url", {})
                    if isinstance(video_url, dict):
                        url = video_url.get("url")
                    else:
                        url = video_url
                    if url:
                        video_data.append(url)
                        result.append({"type": "video"})

        return result

    def _extract_score_from_logprobs(self, ret: Dict[str, Any]) -> float:
        """Extract reranking score from generation response with logprobs."""
        # Get logprobs from the response
        meta_info = ret.get("meta_info", {})
        output_top_logprobs = meta_info.get("output_top_logprobs", [])

        # Use output_top_logprobs[0] - the model's prediction for the first generated token
        top_logprobs = output_top_logprobs[0] if output_top_logprobs else []

        # Find yes and no token probabilities
        # Format: list of tuples (logprob, token_id, token_text)
        p_yes = 0.0
        p_no = 0.0
        found_yes = False
        found_no = False

        for item in top_logprobs:
            logprob, token_id = item[0], item[1]
            if token_id == self._yes_token_id:
                p_yes = math.exp(logprob)
                found_yes = True
            elif token_id == self._no_token_id:
                p_no = math.exp(logprob)
                found_no = True
            if found_yes and found_no:
                break

        return _qwen3_rerank_score(p_yes, p_no)

    def _build_rerank_response(
        self, ret: Union[List[Dict[str, Any]], List[float]], request: V1RerankReqInput
    ) -> List[RerankResponse]:
        """Build the rerank response from generation results"""
        responses = []
        for idx, item in enumerate(ret):
            if isinstance(item, dict):
                score_val = item.get("embedding")
                # Some rerank/reward models return scalar score as embedding[0].
                if isinstance(score_val, list):
                    if len(score_val) == 0 or not isinstance(
                        score_val[0], (int, float)
                    ):
                        raise ValueError(
                            f"Invalid embedding score for rerank at index {idx}: {score_val!r}"
                        )
                    score_val = float(score_val[0])
                responses.append(
                    RerankResponse(
                        score=float(score_val),
                        document=(
                            request.documents[idx] if request.return_documents else None
                        ),
                        index=idx,
                        meta_info=item.get("meta_info"),
                    )
                )
            else:
                responses.append(
                    RerankResponse(
                        score=float(item),
                        document=(
                            request.documents[idx] if request.return_documents else None
                        ),
                        index=idx,
                    )
                )

        # When top_n is set, nlargest avoids fully sorting the candidate list
        # (O(N log top_n) vs O(N log N)) — meaningful for large rerank batches.
        # Validator (V1RerankReqInput.validate_top_n) guarantees top_n >= 1.
        if request.top_n is not None:
            return heapq.nlargest(request.top_n, responses, key=lambda x: x.score)

        responses.sort(key=lambda x: x.score, reverse=True)
        return responses