Renderers and chat templates: from messages to token ids

_VLLM_RENDERERS = {
    "alpamayo2_super": ("hf", "HfRenderer"),
    "cohere": ("cohere", "CohereRenderer"),
    "deepseek_v32": ("deepseek_v32", "DeepseekV32Renderer"),
    "deepseek_v4": ("deepseek_v4", "DeepseekV4Renderer"),
    "deepseek_v41": ("deepseek_v4", "DeepseekV41Renderer"),
    "hf": ("hf", "HfRenderer"),
    "kimi_audio": ("hf", "HfRenderer"),
    "kimi_k3": ("kimi_k3", "KimiK3Renderer"),
    "mistral": ("mistral", "MistralRenderer"),
    "terratorch": ("terratorch", "TerratorchRenderer"),
    "inkling": ("inkling", "InklingRenderer"),
}

def renderer_from_config(config: "VllmConfig", **kwargs):
    model_config = config.model_config

    tokenizer = cached_tokenizer_from_config(model_config, **kwargs)
    renderer_mode, *_ = tokenizer_args_from_config(model_config, **kwargs)

    return RENDERER_REGISTRY.load_renderer(renderer_mode, config, tokenizer)

        # Validate tool_choice when tool parsing is required but unavailable
        if tool_parsing_unavailable and request.tool_choice not in (
            None,
            "none",
        ):
            if request.tool_choice == "auto" and not self.enable_auto_tools:
                # for hf tokenizers, "auto" tools requires
                # --enable-auto-tool-choice and --tool-call-parser
                return self.create_error_response(
                    '"auto" tool choice requires '
                    "--enable-auto-tool-choice and --tool-call-parser to be set"
                )
            elif request.tool_choice != "auto":
                # "required" or named tool requires tool parser
                if isinstance(request.tool_choice, ChatCompletionNamedToolChoiceParam):
                    tool_choice_desc = f'function "{request.tool_choice.function.name}"'
                else:
                    tool_choice_desc = f'"{request.tool_choice}"'
                return self.create_error_response(
                    f"tool_choice={tool_choice_desc} requires "
                    "--tool-call-parser to be set"
                )

        if request.tools is None or (
            request.tool_choice == "none" and self.exclude_tools_when_tool_choice_none
        ):
            tool_dicts = None
        else:
            tool_dicts = [tool.model_dump() for tool in request.tools]

        if not self.use_harmony:
            # Common case.
            error_check_ret = self.validate_chat_template(
                request_chat_template=request.chat_template,
                chat_template_kwargs=request.chat_template_kwargs,
                trust_request_chat_template=self.trust_request_chat_template,
            )
            if error_check_ret is not None:
                return error_check_ret

    async def render_chat_async(
        self,
        conversations: Sequence[list["ChatCompletionMessageParam"]],
        chat_params: ChatParams,
        tok_params: TokenizeParams | None = None,
        *,
        prompt_extras: dict[str, Any] | None = None,
        skip_mm_cache: bool = False,
    ):
        arrival_time = time.time()

        if tok_params is None:
            tok_params = self.default_chat_tok_params

        rendered = [
            self.render_messages_async(conversation, chat_params)
            for conversation in conversations
        ]

        out_conversations = list[list["ConversationMessage"]]()
        dict_prompts = list[DictPrompt]()
        for conv, prompt in await asyncio.gather(*rendered):
            out_conversations.append(conv)
            dict_prompts.append(prompt)

        tok_prompts = await self.tokenize_prompts_async(dict_prompts, tok_params)

        prompt_extras = dict(prompt_extras or {})
        prompt_extras["media_io_kwargs"] = chat_params.media_io_kwargs or {}
        self._apply_prompt_extras(tok_prompts, prompt_extras)

        eng_prompts = await asyncio.gather(
            *(
                self.process_for_engine_async(
                    p, arrival_time, skip_mm_cache=skip_mm_cache, tok_params=tok_params
                )
                for p in tok_prompts
            )
        )

        return out_conversations, eng_prompts

def resolve_chat_template(
    tokenizer: HfTokenizer,
    chat_template: str | None,
    tools: list[dict[str, Any]] | None,
    *,
    model_config: ModelConfig,
) -> str | None:
    # 1st priority: The given chat template
    if chat_template is not None:
        # Resolve template names (e.g. "tool_use") to actual Jinja content
        # so that downstream kwargs detection can parse template variables.
        return tokenizer.get_chat_template(chat_template, tools=tools)

    # 2nd priority: AutoProcessor chat template, unless tool calling is enabled
    if tools is None:
        chat_template = _try_get_processor_chat_template(
            tokenizer,
            revision=model_config.tokenizer_revision,
            code_revision=(
                model_config.code_revision
                if model_config.tokenizer == model_config.model
                else None
            ),
            trust_remote_code=model_config.trust_remote_code,
        )
        if chat_template is not None:
            return chat_template

    # 3rd priority: AutoTokenizer chat template
    try:
        return tokenizer.get_chat_template(chat_template, tools=tools)
    except Exception:
        logger.debug(
            "Failed to load AutoTokenizer chat template for %s",
            tokenizer.name_or_path,
            exc_info=True,
        )

    # 4th priority: Predefined fallbacks
    path = get_chat_template_fallback_path(
        model_type=model_config.hf_config.model_type,
        tokenizer_name_or_path=tokenizer.name_or_path,
    )
    if path is not None:
        logger.info_once(
            "Loading chat template fallback for %s as there isn't one "
            "defined on HF Hub.",
            tokenizer.name_or_path,
        )
        chat_template = load_chat_template(path)
    else:
        logger.debug_once(
            "There is no chat template fallback for %s", tokenizer.name_or_path
        )

    return chat_template

    def build_chat_params(
        self,
        default_template: str | None,
        default_template_content_format: ChatTemplateContentFormatOption,
    ) -> ChatParams:
        extra_kwargs: dict[str, Any] = dict(
            add_generation_prompt=self.add_generation_prompt,
            continue_final_message=self.continue_final_message,
            documents=self.documents,
            reasoning_effort=self.reasoning_effort,
        )

        # When reasoning is requested, activate thinking for models whose
        # chat templates require explicit opt-in (e.g., Gemma4 defaults
        # enable_thinking to false). For templates that don't declare the
        # variable, resolve_chat_template_kwargs filters it out harmlessly.
        user_kwargs = self.chat_template_kwargs or {}
        if self.reasoning_effort is not None and "enable_thinking" not in user_kwargs:
            extra_kwargs["enable_thinking"] = self.reasoning_effort != "none"

        return ChatParams(
            chat_template=self.chat_template or default_template,
            chat_template_content_format=default_template_content_format,
            chat_template_kwargs=merge_kwargs(
                self.chat_template_kwargs,
                extra_kwargs,
            ),
            media_io_kwargs=self.media_io_kwargs,
            # No-tools requests default to tool_choice="none" at the API
            # layer. Collapse that default before rendering, so K3 emits a
            # model-visible tool-choice instruction only for requests with a
            # tools block.
            tool_choice=self.tool_choice if self.tools else None,
            response_format=self.response_format,
        )

        # Thread pool executor for blocking tokenizer operations.  The
        # multimodal processor receives a deep-copied tokenizer (see #36557)
        # so it is safe to run tokenization and MM preprocessing concurrently.
        pool_workers = config.model_config.renderer_num_workers
        self._executor = _SwappableExecutor(max_workers=pool_workers)
        self._resources.callback(self._executor.shutdown, wait=False)

        # Separate single-worker executor so tokenization never queues behind
        # MM preprocessing; must stay single-worker per #38418 (P0/P1 order).
        self._mm_executor = ThreadPoolExecutor(max_workers=1)
        self._resources.callback(self._mm_executor.shutdown, wait=False)

        # Offload tokenization to the thread pool. The sync
        # ``_tokenize_prompt`` already encapsulates the unified ``__call__``
        # path and char-offset extraction, so the async variant is just it
        # offloaded (mirrors ``_process_multimodal_async`` below).
        self._tokenize_prompt_async = make_async(
            self._tokenize_prompt, executor=self._executor
        )

    async def _render_with_timeout(self, coro):
        """Run a template-rendering coroutine with a timeout.

        On timeout, replaces the executor so subsequent renders are not
        permanently blocked by the stuck thread.
        """
        timeout = envs.VLLM_CHAT_TEMPLATE_RENDER_TIMEOUT
        if timeout <= 0:
            return await coro
        try:
            return await asyncio.wait_for(coro, timeout=timeout)
        except TimeoutError:
            self._replace_executor()
            raise TimeoutError(
                f"Chat template rendering timed out after "
                f"{timeout}s. This may indicate a malicious or "
                f"excessively complex template. Adjust "
                f"VLLM_CHAT_TEMPLATE_RENDER_TIMEOUT to override."
            ) from None

    def build_tok_params(self, model_config: ModelConfig) -> TokenizeParams:
        if self.max_completion_tokens is not None:
            max_output_tokens: int | None = self.max_completion_tokens
            max_output_tokens_param = "max_completion_tokens"
        else:
            max_output_tokens = self.max_tokens
            max_output_tokens_param = "max_tokens"

        return TokenizeParams(
            max_total_tokens=model_config.max_model_len,
            max_output_tokens=max_output_tokens,
            truncate_prompt_tokens=self.truncate_prompt_tokens,
            truncation_side=self.truncation_side,
            add_special_tokens=self.add_special_tokens,
            needs_detokenization=bool(self.echo and not self.return_token_ids),
            max_total_tokens_param="max_model_len",
            max_output_tokens_param=max_output_tokens_param,
            return_token_offsets=bool(self.return_token_offsets),
        )

    def _token_len_check(self, tokenizer: TokenizerLike | None, tokens: _S) -> _S:
        """Apply length checks to prompt tokens if necessary."""
        max_input_tokens = self.max_input_tokens
        if max_input_tokens is None:
            return tokens

        if len(tokens) > max_input_tokens:
            token_count = len(tokens)
            # The tokenizer may have truncated the prompt to
            # max_input_tokens + 1 (see get_encode_kwargs), so the
            # actual prompt length could be larger.
            qualifier = "at least " if token_count == max_input_tokens + 1 else ""
            raise VLLMValidationError(
                f"This model's maximum context length is "
                f"{self.max_total_tokens} tokens. However, your prompt "
                f"contains {qualifier}{token_count} input tokens, which "
                f"exceeds this limit even without any output tokens. "
                f"Please reduce the length of the input prompt.",
                parameter="input_tokens",
                value=token_count,
            )