vllm.renderers.online_derenderer

Classes:

• OnlineDerenderer –

OnlineDerenderer

Methods:

• derender_chat_stream –

  Process one GenerateStreamResponse chunk for streaming chat derender.

• derender_completion_stream –

  Process one GenerateStreamResponse chunk for streaming completions.

Source code in vllm/renderers/online_derenderer.py

class OnlineDerenderer:
    def __init__(
        self,
        model_config: ModelConfig,
        renderer: BaseRenderer,
        *,
        request_logger: RequestLogger | None,
        chat_template: str | None,
        chat_template_content_format: ChatTemplateContentFormatOption,
        trust_request_chat_template: bool = False,
        enable_auto_tools: bool = False,
        exclude_tools_when_tool_choice_none: bool = False,
        tool_parser: str | None = None,
        reasoning_parser: str | None = None,
        tool_strict_level: str | None = None,
        default_chat_template_kwargs: dict[str, Any] | None = None,
        log_error_stack: bool = False,
    ) -> None:
        self.model_config = model_config
        self.renderer = renderer
        self.request_logger = request_logger

        self.enable_auto_tools = enable_auto_tools
        self.exclude_tools_when_tool_choice_none = exclude_tools_when_tool_choice_none
        self.use_harmony = model_config.hf_config.model_type == "gpt_oss"
        self.parser: type[Parser] | None = ParserManager.get_parser(
            tool_parser_name=tool_parser,
            reasoning_parser_name=reasoning_parser,
            enable_auto_tools=enable_auto_tools,
            tool_strict_level=tool_strict_level,
            model_name=model_config.model,
            is_harmony=self.use_harmony,
            tokenizer=renderer.tokenizer,
        )

        self.chat_template = chat_template
        self.chat_template_content_format: ChatTemplateContentFormatOption = (
            chat_template_content_format
        )
        self.default_chat_template_kwargs: dict[str, Any] = (
            default_chat_template_kwargs or {}
        )
        self.trust_request_chat_template = trust_request_chat_template

        self.log_error_stack = log_error_stack
        self.supports_browsing = False
        self.supports_code_interpreter = False

        # Detokenization, logprob resolution and parsing are CPU-bound;
        # offload them in one hop to keep the event loop responsive.
        self._derender_chat_async = make_async(
            self._derender_chat, executor=renderer._executor
        )
        self._derender_completion_async = make_async(
            self._derender_completion, executor=renderer._executor
        )
        self._detokenize_delta_async = make_async(
            self._detokenize_delta, executor=renderer._executor
        )
        # Replay is O(n) per chunk, so it must not run on the event loop.
        self._derender_chat_stream_parsed_async = make_async(
            self._derender_chat_stream_parsed, executor=renderer._executor
        )

    async def derender_chat(
        self,
        generate_response: GenerateTokensResponse,
        chat_request: ChatCompletionRequest | None = None,
        prompt_token_ids: list[int] | None = None,
    ) -> list[ChatCompletionResponseChoice]:
        return await self._derender_chat_async(
            generate_response, chat_request, prompt_token_ids
        )

    @cached_property
    def _model_eos_token_ids(self) -> frozenset[int]:
        """EOS ids the engine stops on: the tokenizer's and generation_config's."""
        eos_ids: set[int] = set()
        if (eos_token_id := self.renderer.get_eos_token_id()) is not None:
            eos_ids.add(eos_token_id)
        config_eos = self.model_config.try_get_generation_config().get("eos_token_id")
        if config_eos is not None:
            eos_ids.update([config_eos] if isinstance(config_eos, int) else config_eos)
        return frozenset(eos_ids)

    def _text_token_ids(
        self,
        token_ids: list[int],
        finish_reason: str | None,
        stop_reason: int | str | None,
        request: ChatCompletionRequest | None,
    ) -> list[int]:
        """The ids whose text the engine returns. Like its detokenizer, leave
        out the token that stopped generation (EOS when `stop_reason` is None,
        a `stop_token_ids` entry when it is an int) unless
        `include_stop_str_in_output` is set. The last-id check guards against
        `finish_reason` defaulting to "stop" when the client sends neither."""
        if (
            finish_reason != "stop"
            or isinstance(stop_reason, str)
            or not token_ids
            or (request is not None and request.include_stop_str_in_output)
        ):
            return token_ids
        last = token_ids[-1]
        if stop_reason is None:
            stopped = last in self._model_eos_token_ids
        else:
            stopped = last == stop_reason
        return token_ids[:-1] if stopped else token_ids

    def _derender_chat(
        self,
        generate_response: GenerateTokensResponse,
        chat_request: ChatCompletionRequest | None = None,
        prompt_token_ids: list[int] | None = None,
    ) -> list[ChatCompletionResponseChoice]:
        tokenizer = self.renderer.get_tokenizer()
        choices: list[ChatCompletionResponseChoice] = []

        has_parser = self.parser is not None and chat_request is not None
        skip_special, spaces_between = _decode_params(
            tokenizer, chat_request, preserve_special=has_parser
        )
        seed_ids = (
            prompt_token_ids
            if prompt_token_ids is not None
            else generate_response.prompt_token_ids
        )
        # Choices continue the same prompt, so they share one seeded window.
        seed_state = _seed_stream_state(
            tokenizer,
            seed_ids,
            skip_special_tokens=skip_special,
        )

        chat_template_kwargs = (
            self._resolve_chat_template_kwargs(chat_request)
            if has_parser and chat_request is not None
            else {}
        )
        # Like the coupled chat endpoint, hide per-token metadata (logprobs,
        # token ids) when a parser is configured and reasoning is hidden:
        # either would otherwise leak the reasoning text.
        hide_metadata = (
            has_parser
            and chat_request is not None
            and not chat_request.include_reasoning
        )
        return_token_ids = (
            chat_request is not None
            and chat_request.return_token_ids
            and not hide_metadata
        )
        top_limit = _chat_top_logprobs_limit(chat_request)

        for choice in generate_response.choices:
            if not choice.token_ids:
                raise ValueError(f"choice {choice.index} has empty or null token_ids")

            resolved_logprobs = (
                _resolve_logprobs(
                    choice.logprobs,
                    tokenizer,
                    top_limit=top_limit,
                )
                if choice.logprobs is not None and not hide_metadata
                else None
            )

            # With a parser, special tokens are preserved so it can see
            # markers like </think>, <tool_call>, or Harmony channel tokens.
            # Without one, the request's skip_special_tokens is honoured
            # (default True when no request was given).
            decoded_text, _ = self._detokenize_delta(
                tokenizer,
                self._text_token_ids(
                    choice.token_ids,
                    choice.finish_reason,
                    choice.stop_reason,
                    chat_request,
                ),
                seed_state,
                skip_special_tokens=skip_special,
                spaces_between_special_tokens=spaces_between,
            )

            if has_parser:
                assert chat_request is not None
                message = self._parse_and_assemble(
                    tokenizer,
                    decoded_text,
                    choice.token_ids,
                    chat_request,
                    chat_template_kwargs,
                    generate_response.prompt_token_ids,
                )
            else:
                message = ChatMessage(role="assistant", content=decoded_text)

            choices.append(
                ChatCompletionResponseChoice(
                    index=choice.index,
                    message=message,
                    logprobs=resolved_logprobs,
                    finish_reason=resolve_finish_reason(
                        choice.finish_reason, chat_request, bool(message.tool_calls)
                    ),
                    stop_reason=choice.stop_reason,
                    token_ids=choice.token_ids if return_token_ids else None,
                )
            )

        return choices

    def _resolve_chat_template_kwargs(
        self, chat_request: ChatCompletionRequest
    ) -> dict[str, Any]:
        if self.use_harmony:
            return {}
        return (
            chat_request.build_chat_params(
                self.chat_template,
                self.chat_template_content_format,
            )
            .with_defaults(self.default_chat_template_kwargs)
            .chat_template_kwargs
        )

    def _parse_and_assemble(
        self,
        tokenizer: TokenizerLike,
        text: str,
        token_ids: Sequence[int],
        chat_request: ChatCompletionRequest,
        chat_template_kwargs: dict[str, Any],
        prompt_token_ids: list[int] | None,
    ) -> ChatMessage:
        """Parse decoded output text into an assistant `ChatMessage`.

        Args:
            tokenizer: Tokenizer handed to the parser.
            text: Decoded output text to parse.
            token_ids: Output token IDs the text was decoded from.
            chat_request: Request supplying tools, `tool_choice` and
                `include_reasoning`.
            chat_template_kwargs: Already resolved template kwargs.
            prompt_token_ids: Prompt token IDs, if known.

        Returns:
            The assembled assistant message.

        """
        assert self.parser is not None
        parser = self.parser(
            tokenizer,
            chat_request.tools,
            chat_template_kwargs=chat_template_kwargs,
            model_config=self.model_config,
        )
        if prompt_token_ids is not None:
            parser.set_prompt_token_ids(prompt_token_ids)
        reasoning, content, tool_calls = parser.parse(
            text,
            chat_request,
            enable_auto_tools=self.enable_auto_tools,
            model_output_token_ids=token_ids,
        )

        if not getattr(chat_request, "include_reasoning", True):
            reasoning = None

        tc_items = (
            [
                ToolCall(id=tc.id or make_tool_call_id(), function=tc)
                for tc in tool_calls
            ]
            if tool_calls
            else []
        )

        is_named_tool_choice = (
            type(chat_request.tool_choice) is ChatCompletionNamedToolChoiceParam
        )
        is_required_tool_choice = chat_request.tool_choice == "required"
        if is_named_tool_choice or is_required_tool_choice:
            content = content or ""

        return ChatMessage(
            role="assistant",
            reasoning=reasoning,
            content=content,
            tool_calls=tc_items,
        )

    def _detokenize_delta(
        self,
        tokenizer: TokenizerLike,
        delta_token_ids: list[int],
        state: DerenderStreamState,
        skip_special_tokens: bool = True,
        spaces_between_special_tokens: bool = True,
    ) -> tuple[str, DerenderStreamState]:
        """Incrementally detokenize ``delta_token_ids`` from prior stream state.

        Resumes decoding from the offsets carried in ``state`` rather than
        replaying token history. ``state.prev_tokens`` holds the trailing decode
        window (from ``prefix_offset`` onward) that ``detokenize_incrementally``
        still needs to reproduce any partially read multi-byte character
        (tracked by ``read_offset``). The delta tokens are fed straight onto it.

        The window is bounded. ``detokenize_incrementally`` never reads before
        ``prefix_offset``, so after each token we trim ``prev_tokens`` to that
        tail and rebase the offsets to it. State transport therefore stays
        O(window) per chunk instead of re-sending the full token history.

        Args:
            tokenizer: The tokenizer to decode with.
            delta_token_ids: New token IDs from this generate chunk.
            state: Client carried detok state from the previous call.
            skip_special_tokens: Passed through to the tokenizer.
            spaces_between_special_tokens: Passed through to the tokenizer.

        Returns:
            (new_text, updated_state) — the delta text for this chunk and the
            state to pass to the next call.

        """
        prev_tokens = list(state.prev_tokens)
        prefix_offset = state.prefix_offset
        read_offset = state.read_offset

        text_parts: list[str] = []
        for tok_id in delta_token_ids:
            # prev_tokens is a (possibly empty) list, never None, so this
            # always takes the non first iter path and only consumes
            # all_input_ids[-1].
            new_toks, text, prefix_offset, read_offset = detokenize_incrementally(
                tokenizer=tokenizer,
                all_input_ids=[tok_id],
                prev_tokens=prev_tokens,
                prefix_offset=prefix_offset,
                read_offset=read_offset,
                skip_special_tokens=skip_special_tokens,
                spaces_between_special_tokens=spaces_between_special_tokens,
            )
            # Trim to the tail still readable by detokenize_incrementally
            # (everything before prefix_offset is dead) and rebase the
            # offsets, so the window stays bounded within a long batch
            # decode as well as across chunks.
            prev_tokens = (prev_tokens + new_toks)[prefix_offset:]
            read_offset -= prefix_offset
            prefix_offset = 0
            text_parts.append(text)

        updated_state = state.model_copy(
            update={
                "prev_tokens": prev_tokens,
                "prefix_offset": prefix_offset,
                "read_offset": read_offset,
            }
        )
        return "".join(text_parts), updated_state

    async def derender_chat_stream(
        self,
        model: str,
        generate_chunk: GenerateTokensStreamResponse,
        state: DerenderStreamState | None = None,
        chat_request: ChatCompletionRequest | None = None,
        prompt_tokens: int | None = None,
        prompt_token_ids: list[int] | None = None,
    ) -> tuple[ChatCompletionStreamResponse, DerenderStreamState]:
        """Process one GenerateStreamResponse chunk for streaming chat derender.

        Unlike OpenAI's API, which always emits `role: "assistant"` on the
        very first chunk, this emits it on the first chunk with a non empty
        `choices` list. A leading usage only chunk therefore defers the
        role to the following content chunk instead of sending an empty
        role only delta.

        Args:
            model: Model name for the response object.
            generate_chunk: One SSE chunk from `/inference/v1/generate`.
            state: Client carried detok state (`None` for first call).
            chat_request: Original ChatCompletionRequest from `/render`.
                Required when a reasoning or tool parser is configured
                (validated by the caller — see `ServingDerender`) because
                plain detokenization would leak raw parser markup into `content`.
            prompt_tokens: Prompt token count for the usage chunk.
            prompt_token_ids: Prompt token IDs. Required on the parser path
                (validated by the caller) and optional otherwise, falling
                back to ``generate_chunk.prompt_token_ids``. See
                `DerenderChatStreamRequest.prompt_token_ids`.

        Returns:
            (chunk, updated_state) — the derendered SSE chunk and the state
            the client must pass to the next call.

        """
        # A single DerenderStreamState is threaded through every choice in
        # this chunk. Correct only when there is at most one choice per SSE
        # event (n=1, one call per index), as the streaming derender
        # protocol assumes. Multiple choices sharing one chunk would corrupt
        # each other's detok/parser state.
        if len(generate_chunk.choices) > 1:
            raise ValueError(
                "derender_chat_stream expects at most one choice per chunk"
            )

        parser_cls = self.parser
        if parser_cls is not None:
            # Fail (mirrors ServingDerender's pre-check) because a parser
            # configured model must never fall through to plain detok or
            # reasoning/tool markup would leak into `delta.content`.
            if chat_request is None:
                raise ValueError(
                    "chat_request is required for streaming chat derender "
                    "when a tool or reasoning parser is configured"
                )
            return await self._derender_chat_stream_parsed_async(
                parser_cls,
                model,
                generate_chunk,
                state if state is not None else DerenderStreamState(),
                chat_request,
                prompt_tokens,
                prompt_token_ids,
            )

        tokenizer = self.renderer.get_tokenizer()
        skip_special, spaces_between = _decode_params(tokenizer, chat_request)
        # Seed on the first chunk only. A carried state already has the prompt.
        if state is None:
            state = _seed_stream_state(
                tokenizer,
                prompt_token_ids
                if prompt_token_ids is not None
                else generate_chunk.prompt_token_ids,
                skip_special_tokens=skip_special,
            )
        stream_choices: list[ChatCompletionResponseStreamChoice] = []
        updated_state = state

        return_token_ids = chat_request is not None and chat_request.return_token_ids
        top_limit = _chat_top_logprobs_limit(chat_request)
        for choice in generate_chunk.choices:
            delta_tids = choice.token_ids or []
            new_text, updated_state = await self._detokenize_delta_async(
                tokenizer,
                self._text_token_ids(
                    delta_tids, choice.finish_reason, choice.stop_reason, chat_request
                ),
                updated_state,
                skip_special_tokens=skip_special,
                spaces_between_special_tokens=spaces_between,
            )

            resolved_logprobs = None
            if choice.logprobs is not None:
                resolved_logprobs = _resolve_logprobs(
                    choice.logprobs,
                    tokenizer,
                    initial_context_token_ids=state.logprob_context_token_ids,
                    top_limit=top_limit,
                )

            include_role = not updated_state.role_sent
            updated_state = updated_state.model_copy(
                update={
                    "role_sent": True,
                    "logprob_context_token_ids": _logprob_context_tail(
                        state.logprob_context_token_ids, delta_tids
                    ),
                }
            )

            delta = DeltaMessage(
                role="assistant" if include_role else None,
                content=new_text if new_text else None,
            )
            stream_choices.append(
                ChatCompletionResponseStreamChoice(
                    index=choice.index,
                    delta=delta,
                    logprobs=resolved_logprobs,
                    finish_reason=choice.finish_reason,
                    stop_reason=choice.stop_reason,
                    token_ids=choice.token_ids if return_token_ids else None,
                )
            )

        usage: UsageInfo | None = None
        if generate_chunk.usage is not None:
            u = generate_chunk.usage
            pt = prompt_tokens if prompt_tokens is not None else (u.prompt_tokens or 0)
            ct = u.completion_tokens or 0
            usage = UsageInfo(
                prompt_tokens=pt,
                completion_tokens=ct,
                total_tokens=pt + ct,
            )

        chunk = ChatCompletionStreamResponse(
            id=generate_chunk.request_id,
            model=model,
            choices=stream_choices,
            usage=usage,
            metrics=generate_chunk.metrics,
        )
        return chunk, updated_state

    def _derender_chat_stream_parsed(
        self,
        parser_cls: type[Parser],
        model: str,
        generate_chunk: GenerateTokensStreamResponse,
        state: DerenderStreamState,
        chat_request: ChatCompletionRequest,
        prompt_tokens: int | None,
        prompt_token_ids: list[int] | None,
    ) -> tuple[ChatCompletionStreamResponse, DerenderStreamState]:
        """Parser path for streaming chat derender: replay + `parse_delta`.

        Parser internal state (buffered markup, reasoning/tool phase, etc.)
        cannot be serialized into `DerenderStreamState`, so each call
        builds a fresh parser and replays every prior output token through
        `parse_delta` (discarding the result) before processing this
        chunk's tokens for real.

        Every chunk, replayed or live, goes through one `parse_delta` call
        with the tokens it arrived with (more than one under e.g.
        speculative decoding). Replay recovers those boundaries from
        `state.output_chunk_lens`, so the rebuilt parser state matches
        standard serving, which calls `parse_delta` once per engine step.
        Text comes from a fresh incremental detokenizer with special tokens
        preserved (``skip_special_tokens=False``), seeded from the prompt
        tail on every call since replay starts from scratch.

        Logprobs need no replay: a generate chunk's ``logprobs`` cover exactly
        that chunk's tokens, so they are resolved once and attached to
        whatever delta the parser emits for the chunk, as the coupled chat
        endpoint does. Only the byte-fallback context crosses chunks, via
        ``state.logprob_context_token_ids``. With ``include_reasoning=False``
        logprobs are dropped on every chunk, again like the coupled endpoint.
        """
        tokenizer = self.renderer.get_tokenizer()
        hide_metadata = not chat_request.include_reasoning
        top_limit = _chat_top_logprobs_limit(chat_request)

        parser = parser_cls(
            tokenizer,
            chat_request.tools,
            chat_template_kwargs=self._resolve_chat_template_kwargs(chat_request),
            model_config=self.model_config,
        )

        # Ephemeral incremental detok window, local to this call. Threaded
        # across both the replay and current chunk phases (via `nonlocal`)
        # so multi-byte characters split across that boundary still decode
        # correctly. Discarded once the call returns.
        skip_special, spaces_between = _decode_params(
            tokenizer, chat_request, preserve_special=True
        )
        detok_state = _seed_stream_state(
            tokenizer, prompt_token_ids, skip_special_tokens=skip_special
        )

        def _replay(token_ids: list[int], chunk_lens: list[int]) -> None:
            """Replay prior chunks through `parse_delta` to rebuild parser
            state, discarding the result. Never `finished` since that only
            applies to the current chunk.
            """
            nonlocal detok_state
            start = 0
            for chunk_len in chunk_lens:
                chunk = token_ids[start : start + chunk_len]
                start += chunk_len
                text, detok_state = self._detokenize_delta(
                    tokenizer,
                    chunk,
                    detok_state,
                    skip_special_tokens=skip_special,
                    spaces_between_special_tokens=spaces_between,
                )
                parser.parse_delta(
                    text,
                    chunk,
                    chat_request,
                    prompt_token_ids=prompt_token_ids,
                    finished=False,
                )

        # Replay history to reconstruct parser state. The result is thrown
        # away and only the current chunk's emission goes to the client.
        _replay(state.output_token_ids, state.output_chunk_lens)

        stream_choices: list[ChatCompletionResponseStreamChoice] = []
        output_token_ids = list(state.output_token_ids)
        output_chunk_lens = list(state.output_chunk_lens)
        role_sent = state.role_sent
        tools_streamed = state.tools_streamed
        last_tool_call_ids = list(state.last_tool_call_ids)
        logprob_context = list(state.logprob_context_token_ids)

        return_token_ids = chat_request.return_token_ids and not hide_metadata

        # At most one choice: the caller (derender_chat_stream) already
        # rejects >1 before dispatching here. role_sent/tools_streamed/
        # output_token_ids below are updated for that single choice, not
        # accumulated across choices. Looping over generate_chunk.choices
        # could silently corrupt output if chunks were ever allowed to
        # contain multiple choices.
        if generate_chunk.choices:
            choice = generate_chunk.choices[0]
            delta_tids = choice.token_ids or []
            is_finished = choice.finish_reason is not None

            if delta_tids:
                # One parse_delta call for the whole chunk (producer
                # granularity), not one per token. See the granularity note
                # in this method's docstring.
                text, detok_state = self._detokenize_delta(
                    tokenizer,
                    self._text_token_ids(
                        delta_tids,
                        choice.finish_reason,
                        choice.stop_reason,
                        chat_request,
                    ),
                    detok_state,
                    skip_special_tokens=skip_special,
                    spaces_between_special_tokens=spaces_between,
                )
                delta_message = parser.parse_delta(
                    text,
                    delta_tids,
                    chat_request,
                    prompt_token_ids=prompt_token_ids,
                    finished=is_finished,
                )
            elif is_finished:
                # Finish only chunk (no new tokens). Still flush any
                # buffered tool call arguments.
                delta_message = parser.parse_delta(
                    "",
                    [],
                    chat_request,
                    prompt_token_ids=prompt_token_ids,
                    finished=True,
                )
            else:
                delta_message = None

            if delta_tids:
                output_token_ids.extend(delta_tids)
                output_chunk_lens.append(len(delta_tids))

            resolved_logprobs = None
            if choice.logprobs is not None and not hide_metadata:
                resolved_logprobs = _resolve_logprobs(
                    choice.logprobs,
                    tokenizer,
                    initial_context_token_ids=state.logprob_context_token_ids,
                    top_limit=top_limit,
                )
            logprob_context = _logprob_context_tail(
                state.logprob_context_token_ids, delta_tids
            )

            if delta_message is None:
                delta_message = DeltaMessage()

            if delta_message.tool_calls:
                tools_streamed = True
                for tc in delta_message.tool_calls:
                    if tc.id is None:
                        continue
                    if tc.index < len(last_tool_call_ids):
                        # Pin: reuse the ID already recorded for this index
                        # rather than one a from scratch replay regenerated.
                        tc.id = last_tool_call_ids[tc.index]
                    else:
                        last_tool_call_ids.append(tc.id)

            if not role_sent:
                delta_message.role = "assistant"
                role_sent = True

            finish_reason = choice.finish_reason
            if finish_reason is not None:
                finish_reason = resolve_finish_reason(
                    finish_reason, chat_request, tools_streamed
                )

            stream_choice = ChatCompletionResponseStreamChoice(
                index=choice.index,
                delta=delta_message,
                logprobs=resolved_logprobs,
                finish_reason=finish_reason,
                stop_reason=choice.stop_reason,
                token_ids=choice.token_ids if return_token_ids else None,
            )
            stream_choices.append(
                maybe_filter_parallel_tool_calls(stream_choice, chat_request)
            )

        updated_state = state.model_copy(
            update={
                "output_token_ids": output_token_ids,
                "output_chunk_lens": output_chunk_lens,
                "role_sent": role_sent,
                "tools_streamed": tools_streamed,
                "last_tool_call_ids": last_tool_call_ids,
                "logprob_context_token_ids": logprob_context,
            }
        )

        usage: UsageInfo | None = None
        if generate_chunk.usage is not None:
            u = generate_chunk.usage
            pt = prompt_tokens if prompt_tokens is not None else (u.prompt_tokens or 0)
            ct = u.completion_tokens or 0
            usage = UsageInfo(
                prompt_tokens=pt,
                completion_tokens=ct,
                total_tokens=pt + ct,
            )

        chunk = ChatCompletionStreamResponse(
            id=generate_chunk.request_id,
            model=model,
            choices=stream_choices,
            usage=usage,
            metrics=generate_chunk.metrics,
        )
        return chunk, updated_state

    async def derender_completion(
        self,
        generate_responses: list[GenerateTokensResponse],
        prompt_tokens: list[int] | None = None,
        completion_request: CompletionRequest | None = None,
        prompt_token_ids: list[list[int] | None] | None = None,
    ) -> tuple[list[CompletionResponseChoice], int, int]:
        return await self._derender_completion_async(
            generate_responses, prompt_tokens, completion_request, prompt_token_ids
        )

    def _derender_completion(
        self,
        generate_responses: list[GenerateTokensResponse],
        prompt_tokens: list[int] | None = None,
        completion_request: CompletionRequest | None = None,
        prompt_token_ids: list[list[int] | None] | None = None,
    ) -> tuple[list[CompletionResponseChoice], int, int]:
        n = len(generate_responses)
        prompt_tokens_list: list[int] = (
            prompt_tokens if prompt_tokens is not None else [0] * n
        )
        prompt_token_ids_list: list[list[int] | None] = (
            prompt_token_ids if prompt_token_ids is not None else [None] * n
        )

        tokenizer = self.renderer.get_tokenizer()
        skip_special, spaces_between = _decode_params(tokenizer, completion_request)
        choices: list[CompletionResponseChoice] = []
        total_prompt_tokens = 0
        total_completion_tokens = 0
        index = 0
        top_limit = _completion_top_logprobs_limit(completion_request)
        return_token_ids = (
            completion_request is not None and completion_request.return_token_ids
        )

        for gen, pt, seed_ids_override in zip(
            generate_responses, prompt_tokens_list, prompt_token_ids_list
        ):
            seed_ids = (
                seed_ids_override
                if seed_ids_override is not None
                else gen.prompt_token_ids
            )
            seed_state = _seed_stream_state(
                tokenizer, seed_ids, skip_special_tokens=skip_special
            )
            for choice in gen.choices:
                if not choice.token_ids:
                    raise ValueError(
                        f"choice {choice.index} in response {gen.request_id} "
                        "has empty or null token_ids"
                    )

                decoded_text, _ = self._detokenize_delta(
                    tokenizer,
                    choice.token_ids,
                    seed_state,
                    skip_special_tokens=skip_special,
                    spaces_between_special_tokens=spaces_between,
                )
                completion_logprobs = None
                if choice.logprobs is not None:
                    resolved = _resolve_logprobs(
                        choice.logprobs,
                        tokenizer,
                        top_limit=top_limit,
                    )
                    completion_logprobs = _convert_chat_logprobs_to_completion_logprobs(
                        resolved, choice.logprobs
                    )
                choices.append(
                    CompletionResponseChoice(
                        index=index,
                        text=decoded_text,
                        finish_reason=choice.finish_reason,
                        stop_reason=choice.stop_reason,
                        logprobs=completion_logprobs,
                        token_ids=choice.token_ids if return_token_ids else None,
                    )
                )
                total_completion_tokens += len(choice.token_ids)
                index += 1
            total_prompt_tokens += pt

        return choices, total_prompt_tokens, total_completion_tokens

    async def derender_completion_stream(
        self,
        model: str,
        generate_chunk: GenerateTokensStreamResponse,
        state: DerenderStreamState | None = None,
        prompt_tokens: int | None = None,
        completion_request: CompletionRequest | None = None,
        prompt_token_ids: list[int] | None = None,
    ) -> tuple[CompletionStreamResponse, DerenderStreamState]:
        """Process one GenerateStreamResponse chunk for streaming completions.

        Each call takes one SSE chunk from ``/inference/v1/generate`` plus the
        client carried ``stream_state`` and returns a ``CompletionStreamResponse``
        chunk and the updated state.

        The generate stream emits one choice per SSE event, so this method
        processes one output sequence at a time.  For ``n > 1`` the client
        maintains one ``DerenderStreamState`` per ``choice.index``.

        Args:
            model: Model name for the response object.
            generate_chunk: One SSE chunk from ``/inference/v1/generate``.
            state: Client carried detok state (``None`` → first call).
            prompt_tokens: Prompt token count for usage (from the render step).
            completion_request: Original CompletionRequest from ``/render``;
                supplies ``skip_special_tokens``.
            prompt_token_ids: Seeds the first chunk's decode. Falls back to
                ``generate_chunk.prompt_token_ids``.

        Returns:
            (chunk, updated_state) — the derendered chunk and updated state.

        """
        # See the equivalent check in derender_chat_stream: a single
        # DerenderStreamState is threaded through every choice in this
        # chunk, so more than one choice per chunk would corrupt the
        # detok window across choices.
        if len(generate_chunk.choices) > 1:
            raise ValueError(
                "derender_completion_stream expects at most one choice per chunk"
            )

        tokenizer = self.renderer.get_tokenizer()
        skip_special, spaces_between = _decode_params(tokenizer, completion_request)
        # Seed on the first chunk only. A carried state already has the prompt.
        if state is None:
            state = _seed_stream_state(
                tokenizer,
                prompt_token_ids
                if prompt_token_ids is not None
                else generate_chunk.prompt_token_ids,
                skip_special_tokens=skip_special,
            )
        stream_choices: list[CompletionResponseStreamChoice] = []
        updated_state = state

        top_limit = _completion_top_logprobs_limit(completion_request)
        return_token_ids = (
            completion_request is not None and completion_request.return_token_ids
        )
        for choice in generate_chunk.choices:
            delta_tids = choice.token_ids or []
            new_text, updated_state = await self._detokenize_delta_async(
                tokenizer,
                delta_tids,
                updated_state,
                skip_special_tokens=skip_special,
                spaces_between_special_tokens=spaces_between,
            )

            completion_logprobs = None
            if choice.logprobs is not None:
                resolved = _resolve_logprobs(
                    choice.logprobs,
                    tokenizer,
                    initial_context_token_ids=state.logprob_context_token_ids,
                    top_limit=top_limit,
                )
                completion_logprobs = _convert_chat_logprobs_to_completion_logprobs(
                    resolved,
                    choice.logprobs,
                    initial_text_offset=state.logprob_text_offset,
                )

            updated_state = updated_state.model_copy(
                update={
                    "logprob_context_token_ids": _logprob_context_tail(
                        state.logprob_context_token_ids, delta_tids
                    ),
                    "logprob_text_offset": state.logprob_text_offset + len(new_text),
                }
            )

            stream_choices.append(
                CompletionResponseStreamChoice(
                    index=choice.index,
                    text=new_text,
                    logprobs=completion_logprobs,
                    finish_reason=choice.finish_reason,
                    stop_reason=choice.stop_reason,
                    token_ids=choice.token_ids if return_token_ids else None,
                )
            )

        usage: UsageInfo | None = None
        if generate_chunk.usage is not None:
            u = generate_chunk.usage
            pt = prompt_tokens if prompt_tokens is not None else (u.prompt_tokens or 0)
            ct = u.completion_tokens or 0
            usage = UsageInfo(
                prompt_tokens=pt,
                completion_tokens=ct,
                total_tokens=pt + ct,
            )

        chunk = CompletionStreamResponse(
            id=generate_chunk.request_id,
            model=model,
            choices=stream_choices,
            usage=usage,
            metrics=generate_chunk.metrics,
        )
        return chunk, updated_state

_model_eos_token_ids cached property

EOS ids the engine stops on: the tokenizer's and generation_config's.

_derender_chat_stream_parsed(parser_cls, model, generate_chunk, state, chat_request, prompt_tokens, prompt_token_ids)

Parser path for streaming chat derender: replay + parse_delta.

Parser internal state (buffered markup, reasoning/tool phase, etc.) cannot be serialized into DerenderStreamState, so each call builds a fresh parser and replays every prior output token through parse_delta (discarding the result) before processing this chunk's tokens for real.

Every chunk, replayed or live, goes through one parse_delta call with the tokens it arrived with (more than one under e.g. speculative decoding). Replay recovers those boundaries from state.output_chunk_lens, so the rebuilt parser state matches standard serving, which calls parse_delta once per engine step. Text comes from a fresh incremental detokenizer with special tokens preserved (skip_special_tokens=False), seeded from the prompt tail on every call since replay starts from scratch.

Logprobs need no replay: a generate chunk's logprobs cover exactly that chunk's tokens, so they are resolved once and attached to whatever delta the parser emits for the chunk, as the coupled chat endpoint does. Only the byte-fallback context crosses chunks, via state.logprob_context_token_ids. With include_reasoning=False logprobs are dropped on every chunk, again like the coupled endpoint.

Source code in vllm/renderers/online_derenderer.py

def _derender_chat_stream_parsed(
    self,
    parser_cls: type[Parser],
    model: str,
    generate_chunk: GenerateTokensStreamResponse,
    state: DerenderStreamState,
    chat_request: ChatCompletionRequest,
    prompt_tokens: int | None,
    prompt_token_ids: list[int] | None,
) -> tuple[ChatCompletionStreamResponse, DerenderStreamState]:
    """Parser path for streaming chat derender: replay + `parse_delta`.

    Parser internal state (buffered markup, reasoning/tool phase, etc.)
    cannot be serialized into `DerenderStreamState`, so each call
    builds a fresh parser and replays every prior output token through
    `parse_delta` (discarding the result) before processing this
    chunk's tokens for real.

    Every chunk, replayed or live, goes through one `parse_delta` call
    with the tokens it arrived with (more than one under e.g.
    speculative decoding). Replay recovers those boundaries from
    `state.output_chunk_lens`, so the rebuilt parser state matches
    standard serving, which calls `parse_delta` once per engine step.
    Text comes from a fresh incremental detokenizer with special tokens
    preserved (``skip_special_tokens=False``), seeded from the prompt
    tail on every call since replay starts from scratch.

    Logprobs need no replay: a generate chunk's ``logprobs`` cover exactly
    that chunk's tokens, so they are resolved once and attached to
    whatever delta the parser emits for the chunk, as the coupled chat
    endpoint does. Only the byte-fallback context crosses chunks, via
    ``state.logprob_context_token_ids``. With ``include_reasoning=False``
    logprobs are dropped on every chunk, again like the coupled endpoint.
    """
    tokenizer = self.renderer.get_tokenizer()
    hide_metadata = not chat_request.include_reasoning
    top_limit = _chat_top_logprobs_limit(chat_request)

    parser = parser_cls(
        tokenizer,
        chat_request.tools,
        chat_template_kwargs=self._resolve_chat_template_kwargs(chat_request),
        model_config=self.model_config,
    )

    # Ephemeral incremental detok window, local to this call. Threaded
    # across both the replay and current chunk phases (via `nonlocal`)
    # so multi-byte characters split across that boundary still decode
    # correctly. Discarded once the call returns.
    skip_special, spaces_between = _decode_params(
        tokenizer, chat_request, preserve_special=True
    )
    detok_state = _seed_stream_state(
        tokenizer, prompt_token_ids, skip_special_tokens=skip_special
    )

    def _replay(token_ids: list[int], chunk_lens: list[int]) -> None:
        """Replay prior chunks through `parse_delta` to rebuild parser
        state, discarding the result. Never `finished` since that only
        applies to the current chunk.
        """
        nonlocal detok_state
        start = 0
        for chunk_len in chunk_lens:
            chunk = token_ids[start : start + chunk_len]
            start += chunk_len
            text, detok_state = self._detokenize_delta(
                tokenizer,
                chunk,
                detok_state,
                skip_special_tokens=skip_special,
                spaces_between_special_tokens=spaces_between,
            )
            parser.parse_delta(
                text,
                chunk,
                chat_request,
                prompt_token_ids=prompt_token_ids,
                finished=False,
            )

    # Replay history to reconstruct parser state. The result is thrown
    # away and only the current chunk's emission goes to the client.
    _replay(state.output_token_ids, state.output_chunk_lens)

    stream_choices: list[ChatCompletionResponseStreamChoice] = []
    output_token_ids = list(state.output_token_ids)
    output_chunk_lens = list(state.output_chunk_lens)
    role_sent = state.role_sent
    tools_streamed = state.tools_streamed
    last_tool_call_ids = list(state.last_tool_call_ids)
    logprob_context = list(state.logprob_context_token_ids)

    return_token_ids = chat_request.return_token_ids and not hide_metadata

    # At most one choice: the caller (derender_chat_stream) already
    # rejects >1 before dispatching here. role_sent/tools_streamed/
    # output_token_ids below are updated for that single choice, not
    # accumulated across choices. Looping over generate_chunk.choices
    # could silently corrupt output if chunks were ever allowed to
    # contain multiple choices.
    if generate_chunk.choices:
        choice = generate_chunk.choices[0]
        delta_tids = choice.token_ids or []
        is_finished = choice.finish_reason is not None

        if delta_tids:
            # One parse_delta call for the whole chunk (producer
            # granularity), not one per token. See the granularity note
            # in this method's docstring.
            text, detok_state = self._detokenize_delta(
                tokenizer,
                self._text_token_ids(
                    delta_tids,
                    choice.finish_reason,
                    choice.stop_reason,
                    chat_request,
                ),
                detok_state,
                skip_special_tokens=skip_special,
                spaces_between_special_tokens=spaces_between,
            )
            delta_message = parser.parse_delta(
                text,
                delta_tids,
                chat_request,
                prompt_token_ids=prompt_token_ids,
                finished=is_finished,
            )
        elif is_finished:
            # Finish only chunk (no new tokens). Still flush any
            # buffered tool call arguments.
            delta_message = parser.parse_delta(
                "",
                [],
                chat_request,
                prompt_token_ids=prompt_token_ids,
                finished=True,
            )
        else:
            delta_message = None

        if delta_tids:
            output_token_ids.extend(delta_tids)
            output_chunk_lens.append(len(delta_tids))

        resolved_logprobs = None
        if choice.logprobs is not None and not hide_metadata:
            resolved_logprobs = _resolve_logprobs(
                choice.logprobs,
                tokenizer,
                initial_context_token_ids=state.logprob_context_token_ids,
                top_limit=top_limit,
            )
        logprob_context = _logprob_context_tail(
            state.logprob_context_token_ids, delta_tids
        )

        if delta_message is None:
            delta_message = DeltaMessage()

        if delta_message.tool_calls:
            tools_streamed = True
            for tc in delta_message.tool_calls:
                if tc.id is None:
                    continue
                if tc.index < len(last_tool_call_ids):
                    # Pin: reuse the ID already recorded for this index
                    # rather than one a from scratch replay regenerated.
                    tc.id = last_tool_call_ids[tc.index]
                else:
                    last_tool_call_ids.append(tc.id)

        if not role_sent:
            delta_message.role = "assistant"
            role_sent = True

        finish_reason = choice.finish_reason
        if finish_reason is not None:
            finish_reason = resolve_finish_reason(
                finish_reason, chat_request, tools_streamed
            )

        stream_choice = ChatCompletionResponseStreamChoice(
            index=choice.index,
            delta=delta_message,
            logprobs=resolved_logprobs,
            finish_reason=finish_reason,
            stop_reason=choice.stop_reason,
            token_ids=choice.token_ids if return_token_ids else None,
        )
        stream_choices.append(
            maybe_filter_parallel_tool_calls(stream_choice, chat_request)
        )

    updated_state = state.model_copy(
        update={
            "output_token_ids": output_token_ids,
            "output_chunk_lens": output_chunk_lens,
            "role_sent": role_sent,
            "tools_streamed": tools_streamed,
            "last_tool_call_ids": last_tool_call_ids,
            "logprob_context_token_ids": logprob_context,
        }
    )

    usage: UsageInfo | None = None
    if generate_chunk.usage is not None:
        u = generate_chunk.usage
        pt = prompt_tokens if prompt_tokens is not None else (u.prompt_tokens or 0)
        ct = u.completion_tokens or 0
        usage = UsageInfo(
            prompt_tokens=pt,
            completion_tokens=ct,
            total_tokens=pt + ct,
        )

    chunk = ChatCompletionStreamResponse(
        id=generate_chunk.request_id,
        model=model,
        choices=stream_choices,
        usage=usage,
        metrics=generate_chunk.metrics,
    )
    return chunk, updated_state

_detokenize_delta(tokenizer, delta_token_ids, state, skip_special_tokens=True, spaces_between_special_tokens=True)

Incrementally detokenize delta_token_ids from prior stream state.

Resumes decoding from the offsets carried in state rather than replaying token history. state.prev_tokens holds the trailing decode window (from prefix_offset onward) that detokenize_incrementally still needs to reproduce any partially read multi-byte character (tracked by read_offset). The delta tokens are fed straight onto it.

The window is bounded. detokenize_incrementally never reads before prefix_offset, so after each token we trim prev_tokens to that tail and rebase the offsets to it. State transport therefore stays O(window) per chunk instead of re-sending the full token history.

Parameters:

• tokenizer

  (TokenizerLike) –

  The tokenizer to decode with.

• delta_token_ids

  (list[int]) –

  New token IDs from this generate chunk.

• state

  (DerenderStreamState) –

  Client carried detok state from the previous call.

• skip_special_tokens

  (bool, default: True ) –

  Passed through to the tokenizer.

• spaces_between_special_tokens

  (bool, default: True ) –

  Passed through to the tokenizer.

Returns:

• str –

  (new_text, updated_state) — the delta text for this chunk and the

• DerenderStreamState –

  state to pass to the next call.

Source code in vllm/renderers/online_derenderer.py

def _detokenize_delta(
    self,
    tokenizer: TokenizerLike,
    delta_token_ids: list[int],
    state: DerenderStreamState,
    skip_special_tokens: bool = True,
    spaces_between_special_tokens: bool = True,
) -> tuple[str, DerenderStreamState]:
    """Incrementally detokenize ``delta_token_ids`` from prior stream state.

    Resumes decoding from the offsets carried in ``state`` rather than
    replaying token history. ``state.prev_tokens`` holds the trailing decode
    window (from ``prefix_offset`` onward) that ``detokenize_incrementally``
    still needs to reproduce any partially read multi-byte character
    (tracked by ``read_offset``). The delta tokens are fed straight onto it.

    The window is bounded. ``detokenize_incrementally`` never reads before
    ``prefix_offset``, so after each token we trim ``prev_tokens`` to that
    tail and rebase the offsets to it. State transport therefore stays
    O(window) per chunk instead of re-sending the full token history.

    Args:
        tokenizer: The tokenizer to decode with.
        delta_token_ids: New token IDs from this generate chunk.
        state: Client carried detok state from the previous call.
        skip_special_tokens: Passed through to the tokenizer.
        spaces_between_special_tokens: Passed through to the tokenizer.

    Returns:
        (new_text, updated_state) — the delta text for this chunk and the
        state to pass to the next call.

    """
    prev_tokens = list(state.prev_tokens)
    prefix_offset = state.prefix_offset
    read_offset = state.read_offset

    text_parts: list[str] = []
    for tok_id in delta_token_ids:
        # prev_tokens is a (possibly empty) list, never None, so this
        # always takes the non first iter path and only consumes
        # all_input_ids[-1].
        new_toks, text, prefix_offset, read_offset = detokenize_incrementally(
            tokenizer=tokenizer,
            all_input_ids=[tok_id],
            prev_tokens=prev_tokens,
            prefix_offset=prefix_offset,
            read_offset=read_offset,
            skip_special_tokens=skip_special_tokens,
            spaces_between_special_tokens=spaces_between_special_tokens,
        )
        # Trim to the tail still readable by detokenize_incrementally
        # (everything before prefix_offset is dead) and rebase the
        # offsets, so the window stays bounded within a long batch
        # decode as well as across chunks.
        prev_tokens = (prev_tokens + new_toks)[prefix_offset:]
        read_offset -= prefix_offset
        prefix_offset = 0
        text_parts.append(text)

    updated_state = state.model_copy(
        update={
            "prev_tokens": prev_tokens,
            "prefix_offset": prefix_offset,
            "read_offset": read_offset,
        }
    )
    return "".join(text_parts), updated_state

_parse_and_assemble(tokenizer, text, token_ids, chat_request, chat_template_kwargs, prompt_token_ids)

Parse decoded output text into an assistant ChatMessage.

Parameters:

• tokenizer

  (TokenizerLike) –

  Tokenizer handed to the parser.

• text

  (str) –

  Decoded output text to parse.

• token_ids

  (Sequence[int]) –

  Output token IDs the text was decoded from.

• chat_request

  (ChatCompletionRequest) –

  Request supplying tools, tool_choice and include_reasoning.

• chat_template_kwargs

  (dict[str, Any]) –

  Already resolved template kwargs.

• prompt_token_ids

  (list[int] | None) –

  Prompt token IDs, if known.

Returns:

• ChatMessage –

  The assembled assistant message.

Source code in vllm/renderers/online_derenderer.py

def _parse_and_assemble(
    self,
    tokenizer: TokenizerLike,
    text: str,
    token_ids: Sequence[int],
    chat_request: ChatCompletionRequest,
    chat_template_kwargs: dict[str, Any],
    prompt_token_ids: list[int] | None,
) -> ChatMessage:
    """Parse decoded output text into an assistant `ChatMessage`.

    Args:
        tokenizer: Tokenizer handed to the parser.
        text: Decoded output text to parse.
        token_ids: Output token IDs the text was decoded from.
        chat_request: Request supplying tools, `tool_choice` and
            `include_reasoning`.
        chat_template_kwargs: Already resolved template kwargs.
        prompt_token_ids: Prompt token IDs, if known.

    Returns:
        The assembled assistant message.

    """
    assert self.parser is not None
    parser = self.parser(
        tokenizer,
        chat_request.tools,
        chat_template_kwargs=chat_template_kwargs,
        model_config=self.model_config,
    )
    if prompt_token_ids is not None:
        parser.set_prompt_token_ids(prompt_token_ids)
    reasoning, content, tool_calls = parser.parse(
        text,
        chat_request,
        enable_auto_tools=self.enable_auto_tools,
        model_output_token_ids=token_ids,
    )

    if not getattr(chat_request, "include_reasoning", True):
        reasoning = None

    tc_items = (
        [
            ToolCall(id=tc.id or make_tool_call_id(), function=tc)
            for tc in tool_calls
        ]
        if tool_calls
        else []
    )

    is_named_tool_choice = (
        type(chat_request.tool_choice) is ChatCompletionNamedToolChoiceParam
    )
    is_required_tool_choice = chat_request.tool_choice == "required"
    if is_named_tool_choice or is_required_tool_choice:
        content = content or ""

    return ChatMessage(
        role="assistant",
        reasoning=reasoning,
        content=content,
        tool_calls=tc_items,
    )

_text_token_ids(token_ids, finish_reason, stop_reason, request)

The ids whose text the engine returns. Like its detokenizer, leave out the token that stopped generation (EOS when stop_reason is None, a stop_token_ids entry when it is an int) unless include_stop_str_in_output is set. The last-id check guards against finish_reason defaulting to "stop" when the client sends neither.

Source code in vllm/renderers/online_derenderer.py

def _text_token_ids(
    self,
    token_ids: list[int],
    finish_reason: str | None,
    stop_reason: int | str | None,
    request: ChatCompletionRequest | None,
) -> list[int]:
    """The ids whose text the engine returns. Like its detokenizer, leave
    out the token that stopped generation (EOS when `stop_reason` is None,
    a `stop_token_ids` entry when it is an int) unless
    `include_stop_str_in_output` is set. The last-id check guards against
    `finish_reason` defaulting to "stop" when the client sends neither."""
    if (
        finish_reason != "stop"
        or isinstance(stop_reason, str)
        or not token_ids
        or (request is not None and request.include_stop_str_in_output)
    ):
        return token_ids
    last = token_ids[-1]
    if stop_reason is None:
        stopped = last in self._model_eos_token_ids
    else:
        stopped = last == stop_reason
    return token_ids[:-1] if stopped else token_ids

derender_chat_stream(model, generate_chunk, state=None, chat_request=None, prompt_tokens=None, prompt_token_ids=None) async

Process one GenerateStreamResponse chunk for streaming chat derender.

Unlike OpenAI's API, which always emits role: "assistant" on the very first chunk, this emits it on the first chunk with a non empty choices list. A leading usage only chunk therefore defers the role to the following content chunk instead of sending an empty role only delta.

Parameters:

• model

  (str) –

  Model name for the response object.

• generate_chunk

  (GenerateTokensStreamResponse) –

  One SSE chunk from /inference/v1/generate.

• state

  (DerenderStreamState | None, default: None ) –

  Client carried detok state (None for first call).

• chat_request

  (ChatCompletionRequest | None, default: None ) –

  Original ChatCompletionRequest from /render. Required when a reasoning or tool parser is configured (validated by the caller — see ServingDerender) because plain detokenization would leak raw parser markup into content.

• prompt_tokens

  (int | None, default: None ) –

  Prompt token count for the usage chunk.

• prompt_token_ids

  (list[int] | None, default: None ) –

  Prompt token IDs. Required on the parser path (validated by the caller) and optional otherwise, falling back to generate_chunk.prompt_token_ids. See DerenderChatStreamRequest.prompt_token_ids.

Returns:

• ChatCompletionStreamResponse –

  (chunk, updated_state) — the derendered SSE chunk and the state

• DerenderStreamState –

  the client must pass to the next call.

Source code in vllm/renderers/online_derenderer.py

async def derender_chat_stream(
    self,
    model: str,
    generate_chunk: GenerateTokensStreamResponse,
    state: DerenderStreamState | None = None,
    chat_request: ChatCompletionRequest | None = None,
    prompt_tokens: int | None = None,
    prompt_token_ids: list[int] | None = None,
) -> tuple[ChatCompletionStreamResponse, DerenderStreamState]:
    """Process one GenerateStreamResponse chunk for streaming chat derender.

    Unlike OpenAI's API, which always emits `role: "assistant"` on the
    very first chunk, this emits it on the first chunk with a non empty
    `choices` list. A leading usage only chunk therefore defers the
    role to the following content chunk instead of sending an empty
    role only delta.

    Args:
        model: Model name for the response object.
        generate_chunk: One SSE chunk from `/inference/v1/generate`.
        state: Client carried detok state (`None` for first call).
        chat_request: Original ChatCompletionRequest from `/render`.
            Required when a reasoning or tool parser is configured
            (validated by the caller — see `ServingDerender`) because
            plain detokenization would leak raw parser markup into `content`.
        prompt_tokens: Prompt token count for the usage chunk.
        prompt_token_ids: Prompt token IDs. Required on the parser path
            (validated by the caller) and optional otherwise, falling
            back to ``generate_chunk.prompt_token_ids``. See
            `DerenderChatStreamRequest.prompt_token_ids`.

    Returns:
        (chunk, updated_state) — the derendered SSE chunk and the state
        the client must pass to the next call.

    """
    # A single DerenderStreamState is threaded through every choice in
    # this chunk. Correct only when there is at most one choice per SSE
    # event (n=1, one call per index), as the streaming derender
    # protocol assumes. Multiple choices sharing one chunk would corrupt
    # each other's detok/parser state.
    if len(generate_chunk.choices) > 1:
        raise ValueError(
            "derender_chat_stream expects at most one choice per chunk"
        )

    parser_cls = self.parser
    if parser_cls is not None:
        # Fail (mirrors ServingDerender's pre-check) because a parser
        # configured model must never fall through to plain detok or
        # reasoning/tool markup would leak into `delta.content`.
        if chat_request is None:
            raise ValueError(
                "chat_request is required for streaming chat derender "
                "when a tool or reasoning parser is configured"
            )
        return await self._derender_chat_stream_parsed_async(
            parser_cls,
            model,
            generate_chunk,
            state if state is not None else DerenderStreamState(),
            chat_request,
            prompt_tokens,
            prompt_token_ids,
        )

    tokenizer = self.renderer.get_tokenizer()
    skip_special, spaces_between = _decode_params(tokenizer, chat_request)
    # Seed on the first chunk only. A carried state already has the prompt.
    if state is None:
        state = _seed_stream_state(
            tokenizer,
            prompt_token_ids
            if prompt_token_ids is not None
            else generate_chunk.prompt_token_ids,
            skip_special_tokens=skip_special,
        )
    stream_choices: list[ChatCompletionResponseStreamChoice] = []
    updated_state = state

    return_token_ids = chat_request is not None and chat_request.return_token_ids
    top_limit = _chat_top_logprobs_limit(chat_request)
    for choice in generate_chunk.choices:
        delta_tids = choice.token_ids or []
        new_text, updated_state = await self._detokenize_delta_async(
            tokenizer,
            self._text_token_ids(
                delta_tids, choice.finish_reason, choice.stop_reason, chat_request
            ),
            updated_state,
            skip_special_tokens=skip_special,
            spaces_between_special_tokens=spaces_between,
        )

        resolved_logprobs = None
        if choice.logprobs is not None:
            resolved_logprobs = _resolve_logprobs(
                choice.logprobs,
                tokenizer,
                initial_context_token_ids=state.logprob_context_token_ids,
                top_limit=top_limit,
            )

        include_role = not updated_state.role_sent
        updated_state = updated_state.model_copy(
            update={
                "role_sent": True,
                "logprob_context_token_ids": _logprob_context_tail(
                    state.logprob_context_token_ids, delta_tids
                ),
            }
        )

        delta = DeltaMessage(
            role="assistant" if include_role else None,
            content=new_text if new_text else None,
        )
        stream_choices.append(
            ChatCompletionResponseStreamChoice(
                index=choice.index,
                delta=delta,
                logprobs=resolved_logprobs,
                finish_reason=choice.finish_reason,
                stop_reason=choice.stop_reason,
                token_ids=choice.token_ids if return_token_ids else None,
            )
        )

    usage: UsageInfo | None = None
    if generate_chunk.usage is not None:
        u = generate_chunk.usage
        pt = prompt_tokens if prompt_tokens is not None else (u.prompt_tokens or 0)
        ct = u.completion_tokens or 0
        usage = UsageInfo(
            prompt_tokens=pt,
            completion_tokens=ct,
            total_tokens=pt + ct,
        )

    chunk = ChatCompletionStreamResponse(
        id=generate_chunk.request_id,
        model=model,
        choices=stream_choices,
        usage=usage,
        metrics=generate_chunk.metrics,
    )
    return chunk, updated_state

derender_completion_stream(model, generate_chunk, state=None, prompt_tokens=None, completion_request=None, prompt_token_ids=None) async

Process one GenerateStreamResponse chunk for streaming completions.

Each call takes one SSE chunk from /inference/v1/generate plus the client carried stream_state and returns a CompletionStreamResponse chunk and the updated state.

The generate stream emits one choice per SSE event, so this method processes one output sequence at a time. For n > 1 the client maintains one DerenderStreamState per choice.index.

Parameters:

• model

  (str) –

  Model name for the response object.

• generate_chunk

  (GenerateTokensStreamResponse) –

  One SSE chunk from /inference/v1/generate.

• state

  (DerenderStreamState | None, default: None ) –

  Client carried detok state (None → first call).

• prompt_tokens

  (int | None, default: None ) –

  Prompt token count for usage (from the render step).

• completion_request

  (CompletionRequest | None, default: None ) –

  Original CompletionRequest from /render; supplies skip_special_tokens.

• prompt_token_ids

  (list[int] | None, default: None ) –

  Seeds the first chunk's decode. Falls back to generate_chunk.prompt_token_ids.

Returns:

• tuple[CompletionStreamResponse, DerenderStreamState] –

  (chunk, updated_state) — the derendered chunk and updated state.

Source code in vllm/renderers/online_derenderer.py

async def derender_completion_stream(
    self,
    model: str,
    generate_chunk: GenerateTokensStreamResponse,
    state: DerenderStreamState | None = None,
    prompt_tokens: int | None = None,
    completion_request: CompletionRequest | None = None,
    prompt_token_ids: list[int] | None = None,
) -> tuple[CompletionStreamResponse, DerenderStreamState]:
    """Process one GenerateStreamResponse chunk for streaming completions.

    Each call takes one SSE chunk from ``/inference/v1/generate`` plus the
    client carried ``stream_state`` and returns a ``CompletionStreamResponse``
    chunk and the updated state.

    The generate stream emits one choice per SSE event, so this method
    processes one output sequence at a time.  For ``n > 1`` the client
    maintains one ``DerenderStreamState`` per ``choice.index``.

    Args:
        model: Model name for the response object.
        generate_chunk: One SSE chunk from ``/inference/v1/generate``.
        state: Client carried detok state (``None`` → first call).
        prompt_tokens: Prompt token count for usage (from the render step).
        completion_request: Original CompletionRequest from ``/render``;
            supplies ``skip_special_tokens``.
        prompt_token_ids: Seeds the first chunk's decode. Falls back to
            ``generate_chunk.prompt_token_ids``.

    Returns:
        (chunk, updated_state) — the derendered chunk and updated state.

    """
    # See the equivalent check in derender_chat_stream: a single
    # DerenderStreamState is threaded through every choice in this
    # chunk, so more than one choice per chunk would corrupt the
    # detok window across choices.
    if len(generate_chunk.choices) > 1:
        raise ValueError(
            "derender_completion_stream expects at most one choice per chunk"
        )

    tokenizer = self.renderer.get_tokenizer()
    skip_special, spaces_between = _decode_params(tokenizer, completion_request)
    # Seed on the first chunk only. A carried state already has the prompt.
    if state is None:
        state = _seed_stream_state(
            tokenizer,
            prompt_token_ids
            if prompt_token_ids is not None
            else generate_chunk.prompt_token_ids,
            skip_special_tokens=skip_special,
        )
    stream_choices: list[CompletionResponseStreamChoice] = []
    updated_state = state

    top_limit = _completion_top_logprobs_limit(completion_request)
    return_token_ids = (
        completion_request is not None and completion_request.return_token_ids
    )
    for choice in generate_chunk.choices:
        delta_tids = choice.token_ids or []
        new_text, updated_state = await self._detokenize_delta_async(
            tokenizer,
            delta_tids,
            updated_state,
            skip_special_tokens=skip_special,
            spaces_between_special_tokens=spaces_between,
        )

        completion_logprobs = None
        if choice.logprobs is not None:
            resolved = _resolve_logprobs(
                choice.logprobs,
                tokenizer,
                initial_context_token_ids=state.logprob_context_token_ids,
                top_limit=top_limit,
            )
            completion_logprobs = _convert_chat_logprobs_to_completion_logprobs(
                resolved,
                choice.logprobs,
                initial_text_offset=state.logprob_text_offset,
            )

        updated_state = updated_state.model_copy(
            update={
                "logprob_context_token_ids": _logprob_context_tail(
                    state.logprob_context_token_ids, delta_tids
                ),
                "logprob_text_offset": state.logprob_text_offset + len(new_text),
            }
        )

        stream_choices.append(
            CompletionResponseStreamChoice(
                index=choice.index,
                text=new_text,
                logprobs=completion_logprobs,
                finish_reason=choice.finish_reason,
                stop_reason=choice.stop_reason,
                token_ids=choice.token_ids if return_token_ids else None,
            )
        )

    usage: UsageInfo | None = None
    if generate_chunk.usage is not None:
        u = generate_chunk.usage
        pt = prompt_tokens if prompt_tokens is not None else (u.prompt_tokens or 0)
        ct = u.completion_tokens or 0
        usage = UsageInfo(
            prompt_tokens=pt,
            completion_tokens=ct,
            total_tokens=pt + ct,
        )

    chunk = CompletionStreamResponse(
        id=generate_chunk.request_id,
        model=model,
        choices=stream_choices,
        usage=usage,
        metrics=generate_chunk.metrics,
    )
    return chunk, updated_state

_chat_top_logprobs_limit(chat_request)

The cut /v1/chat/completions applies (chat_top_logprobs_limit, the same helper its _get_top_logprobs uses). None keeps every candidate, which is also what happens without a chat_request or with one that does not ask for logprobs: it then says nothing about the cut, so the candidates generate returned are passed through.

Source code in vllm/renderers/online_derenderer.py

def _chat_top_logprobs_limit(chat_request: ChatCompletionRequest | None) -> int | None:
    """The cut `/v1/chat/completions` applies (`chat_top_logprobs_limit`, the
    same helper its `_get_top_logprobs` uses). None keeps every candidate, which
    is also what happens without a `chat_request` or with one that does not ask
    for logprobs: it then says nothing about the cut, so the candidates
    generate returned are passed through."""
    if chat_request is None or not chat_request.logprobs:
        return None
    return chat_top_logprobs_limit(
        chat_request.top_logprobs, bool(chat_request.logprob_token_ids)
    )

_completion_top_logprobs_limit(completion_request)

The cut /v1/completions applies (completion_top_logprobs_limit, the same helper _create_completion_logprobs uses). None keeps every candidate, which is also what happens without a completion_request or with one that does not ask for logprobs (logprobs unset).

Source code in vllm/renderers/online_derenderer.py

def _completion_top_logprobs_limit(
    completion_request: CompletionRequest | None,
) -> int | None:
    """The cut `/v1/completions` applies (`completion_top_logprobs_limit`, the
    same helper `_create_completion_logprobs` uses). None keeps every candidate,
    which is also what happens without a `completion_request` or with one that
    does not ask for logprobs (`logprobs` unset)."""
    if completion_request is None:
        return None
    return completion_top_logprobs_limit(
        completion_request.logprobs, bool(completion_request.logprob_token_ids)
    )

_convert_chat_logprobs_to_completion_logprobs(logprobs, generate_logprobs, initial_text_offset=0)

Convert ChatCompletionLogProbs (per-token objects) to CompletionLogProbs (parallel flat lists) as required by the /v1/completions response schema.

generate_logprobs is what logprobs was resolved from. A position with no candidates there had no logprobs in the engine output, which /v1/completions reports as None; every other position is a dict of the candidates the cut kept.

initial_text_offset keeps text_offset absolute across streaming chunks, mirroring the generate streaming path.

Source code in vllm/renderers/online_derenderer.py

def _convert_chat_logprobs_to_completion_logprobs(
    logprobs: ChatCompletionLogProbs,
    generate_logprobs: GenerateLogProbs,
    initial_text_offset: int = 0,
) -> CompletionLogProbs:
    """Convert ChatCompletionLogProbs (per-token objects) to CompletionLogProbs
    (parallel flat lists) as required by the /v1/completions response schema.

    ``generate_logprobs`` is what ``logprobs`` was resolved from. A position
    with no candidates there had no logprobs in the engine output, which
    /v1/completions reports as ``None``; every other position is a dict of
    the candidates the cut kept.

    ``initial_text_offset`` keeps ``text_offset`` absolute across streaming
    chunks, mirroring the generate streaming path.
    """
    if logprobs.content is None:
        return CompletionLogProbs()

    tokens: list[str] = []
    token_logprobs: list[float | None] = []
    top_logprobs_list: list[dict[str, float] | None] = []
    text_offset: list[int] = []

    offset = initial_text_offset
    for entry, source in zip(logprobs.content, generate_logprobs.content or ()):
        text_offset.append(offset)
        tokens.append(entry.token)
        token_logprobs.append(entry.logprob)
        top_logprobs_list.append(
            {t.token: t.logprob for t in entry.top_logprobs}
            if source.top_logprobs
            else None
        )
        offset += len(entry.token)

    return CompletionLogProbs(
        text_offset=text_offset,
        token_logprobs=token_logprobs,
        tokens=tokens,
        top_logprobs=top_logprobs_list,
    )

_correct_decoded_token(token_id, context_token_ids, tokenizer)

Use preceding tokens as context to fix U+FFFD from byte-fallback.

Mirrors LogprobsProcessor._correct_decoded_token in v1/engine/logprobs.py.

Source code in vllm/renderers/online_derenderer.py

def _correct_decoded_token(
    token_id: int, context_token_ids: list[int], tokenizer: TokenizerLike
) -> str:
    """Use preceding tokens as context to fix U+FFFD from byte-fallback.

    Mirrors LogprobsProcessor._correct_decoded_token in v1/engine/logprobs.py.
    """
    max_ctx = min(len(context_token_ids), 4)

    for num_ctx in range(1, max_ctx + 1):
        context = context_token_ids[-num_ctx:]
        full_decoded = tokenizer.decode(context + [token_id])

        if full_decoded.endswith("�"):
            continue

        clean_end = len(context)
        for j in range(len(context) - 1, -1, -1):
            if tokenizer.decode([context[j]]).endswith("�"):
                clean_end = j
            else:
                break

        clean_prefix = tokenizer.decode(context[:clean_end]) if clean_end > 0 else ""

        if full_decoded.startswith(clean_prefix):
            return full_decoded[len(clean_prefix) :]

        common_len = 0
        for a, b in zip(clean_prefix, full_decoded):
            if a != b:
                break
            common_len += 1
        return full_decoded[common_len:]

    return ""

_decode_params(tokenizer, request, preserve_special=False)

Derive (skip_special_tokens, spaces_between_special_tokens) the way the engine does (IncrementalDetokenizer).

Parameters:

• tokenizer

  (TokenizerLike) –

  Picks the engine's fast or slow detokenizer rules.

• request

  (ChatCompletionRequest | CompletionRequest | None) –

  The original request, if the caller supplied one.

• preserve_special

  (bool, default: False ) –

  Keep special tokens so a parser can see markers. The serving side does this via adjust_request.

Source code in vllm/renderers/online_derenderer.py

def _decode_params(
    tokenizer: TokenizerLike,
    request: ChatCompletionRequest | CompletionRequest | None,
    preserve_special: bool = False,
) -> tuple[bool, bool]:
    """Derive `(skip_special_tokens, spaces_between_special_tokens)` the way
    the engine does (`IncrementalDetokenizer`).

    Args:
        tokenizer: Picks the engine's fast or slow detokenizer rules.
        request: The original request, if the caller supplied one.
        preserve_special: Keep special tokens so a parser can see markers.
            The serving side does this via `adjust_request`.

    """
    skip_special = (
        False
        if preserve_special
        else (request.skip_special_tokens if request is not None else True)
    )
    spaces_between = (
        request.spaces_between_special_tokens if request is not None else True
    )
    # Only the fast detokenizer forces spaces on when skipping special tokens.
    if skip_special and uses_fast_detokenizer(tokenizer):
        spaces_between = True
    return skip_special, spaces_between

_logprob_context_tail(context_token_ids, delta_token_ids)

Advance the carried logprob context by this chunk's sampled tokens.

Source code in vllm/renderers/online_derenderer.py

def _logprob_context_tail(
    context_token_ids: list[int], delta_token_ids: list[int]
) -> list[int]:
    """Advance the carried logprob context by this chunk's sampled tokens."""
    return (list(context_token_ids) + list(delta_token_ids))[-_LOGPROB_CONTEXT_WINDOW:]

_resolve_logprobs(logprobs, tokenizer, initial_context_token_ids=(), top_limit=None)

Convert generate's integer-id logprobs to the OpenAI chat shape.

The generate server has no tokenizer, so token and bytes are filled here, with the same U+FFFD byte-fallback correction the coupled path applies (_correct_decoded_token, which needs the preceding sampled token ids as context). initial_context_token_ids seeds that context with sampled IDs from preceding chunks (streaming), so multi-byte characters split across chunk boundaries still resolve.

Generate returns every candidate the engine produced for a position (the sampled token first, then the top k). top_limit applies the target endpoint's cut before decoding, so each derender endpoint returns what its coupled counterpart would; None keeps all of them.

Source code in vllm/renderers/online_derenderer.py

def _resolve_logprobs(
    logprobs: GenerateLogProbs,
    tokenizer: TokenizerLike,
    initial_context_token_ids: Sequence[int] = (),
    top_limit: int | None = None,
) -> ChatCompletionLogProbs:
    """Convert generate's integer-id logprobs to the OpenAI chat shape.

    The generate server has no tokenizer, so `token` and `bytes` are filled
    here, with the same U+FFFD byte-fallback correction the coupled path
    applies (`_correct_decoded_token`, which needs the preceding sampled token
    ids as context). ``initial_context_token_ids`` seeds that context with
    sampled IDs from preceding chunks (streaming), so multi-byte characters
    split across chunk boundaries still resolve.

    Generate returns every candidate the engine produced for a position (the
    sampled token first, then the top k). ``top_limit`` applies the target
    endpoint's cut before decoding, so each derender endpoint returns what its
    coupled counterpart would; None keeps all of them.
    """
    if logprobs.content is None:
        return ChatCompletionLogProbs()

    context_token_ids: list[int] = list(initial_context_token_ids)
    resolved_content = []

    for entry in logprobs.content:
        top_entries = (
            entry.top_logprobs if top_limit is None else entry.top_logprobs[:top_limit]
        )
        # One batch per position: the sampled id and its kept top-k ids.
        (token_str, token_bytes), *top_decoded = decode_token_ids(
            [entry.token_id, *(top.token_id for top in top_entries)],
            tokenizer,
        )

        if token_str.endswith("\ufffd"):
            token_str = _correct_decoded_token(
                entry.token_id, context_token_ids, tokenizer
            )
            token_bytes = list(token_str.encode("utf-8"))

        resolved_top = []
        for top, (top_str, top_bytes) in zip(top_entries, top_decoded):
            if top_str.endswith("\ufffd"):
                top_str = _correct_decoded_token(
                    top.token_id, context_token_ids, tokenizer
                )
                top_bytes = list(top_str.encode("utf-8"))
            resolved_top.append(
                ChatCompletionLogProb(
                    token=top_str,
                    logprob=top.logprob,
                    bytes=top_bytes,
                )
            )

        resolved_content.append(
            ChatCompletionLogProbsContent(
                token=token_str,
                logprob=entry.logprob,
                bytes=token_bytes,
                top_logprobs=resolved_top,
            )
        )

        context_token_ids.append(entry.token_id)

    return ChatCompletionLogProbs(content=resolved_content)

_seed_stream_state(tokenizer, prompt_token_ids, skip_special_tokens)

Build the initial decode state from the prompt tail, the same way the engine primes its incremental detokenizer. Without it, Metaspace tokenizers drop the first output token's leading space.

Returns an empty state when prompt_token_ids is empty or omitted.

Source code in vllm/renderers/online_derenderer.py

def _seed_stream_state(
    tokenizer: TokenizerLike,
    prompt_token_ids: list[int] | None,
    skip_special_tokens: bool,
) -> DerenderStreamState:
    """Build the initial decode state from the prompt tail, the same way the
    engine primes its incremental detokenizer. Without it, Metaspace
    tokenizers drop the first output token's leading space.

    Returns an empty state when `prompt_token_ids` is empty or omitted.
    """
    if not prompt_token_ids:
        if prompt_token_ids is None and get_leading_space_marker(tokenizer) is not None:
            logger.warning_once(
                "derender got no prompt_token_ids, so the first output token "
                "may lose its leading space compared to the coupled endpoint. "
                "Pass prompt_token_ids from the render step to avoid this."
            )
        return DerenderStreamState()

    prev_tokens, prefix_offset, read_offset = convert_prompt_ids_to_tokens(
        tokenizer, prompt_token_ids, skip_special_tokens=skip_special_tokens
    )
    return DerenderStreamState(
        prev_tokens=prev_tokens,
        prefix_offset=prefix_offset,
        read_offset=read_offset,
    )