Skip to content

vllm.entrypoints.openai.chat_completion.stream_processor

Classes:

ChatStreamProcessor

Turn one chat choice generated chunks into stream deltas.

Shared by OpenAIServingChat streaming and streaming chat derender, so both emit the same reasoning / content / tool_calls deltas and finish_reason for the same chunks. Use one instance per choice and call step once per chunk, in order and with the chunking the engine produced.

Methods:

  • step –

    Parse one chunk and resolve its chat finish_reason.

Source code in vllm/entrypoints/openai/chat_completion/stream_processor.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
class ChatStreamProcessor:
    """Turn one chat choice generated chunks into stream deltas.

    Shared by `OpenAIServingChat` streaming and streaming chat derender, so
    both emit the same `reasoning` / `content` / `tool_calls` deltas and
    `finish_reason` for the same chunks. Use one instance per choice and call
    `step` once per chunk, in order and with the chunking the engine produced.
    """

    def __init__(
        self,
        request: ChatCompletionRequest,
        parser: Parser | None,
        *,
        enable_auto_tools: bool,
        parser_cls: type[Parser] | None,
    ) -> None:
        self.request = request
        self.parser = parser
        self.keep_tool_calls = keeps_tool_calls(
            request, enable_auto_tools=enable_auto_tools, parser_cls=parser_cls
        )
        self.num_output_tokens = 0
        self.tools_streamed = False
        # Logprobs and token IDs would leak hidden reasoning through decoded
        # logprob tokens or raw IDs, so callers drop both on every chunk.
        self.hide_metadata = parser is not None and not request.include_reasoning

    def step(
        self,
        delta_text: str,
        delta_token_ids: list[int],
        finish_reason: str | None,
        prompt_token_ids: list[int] | None,
    ) -> ChatStreamStep:
        """Parse one chunk and resolve its chat `finish_reason`.

        Args:
            delta_text: Text decoded from this chunk's tokens.
            delta_token_ids: This chunk's token IDs.
            finish_reason: The engine finish reason. None if not finished.
            prompt_token_ids: Prompt token IDs if known.

        Returns:
            The delta to stream, filtered for `tool_choice` and
            `parallel_tool_calls`, and the chat `finish_reason`.

        """
        delta: DeltaMessage | None
        if not delta_text and not delta_token_ids and not self.num_output_tokens:
            # Finished without output. Send the final chunk unparsed.
            delta = DeltaMessage()
        elif self.parser is not None:
            delta = self.parser.parse_delta(
                delta_text=delta_text,
                delta_token_ids=delta_token_ids,
                request=self.request,
                prompt_token_ids=prompt_token_ids,
                finished=finish_reason is not None,
            )
            if delta is not None and delta.tool_calls:
                if self.keep_tool_calls:
                    self.tools_streamed = True
                else:
                    delta.tool_calls = []
                    if not delta.content and not delta.reasoning:
                        delta = None
        else:
            delta = DeltaMessage(content=delta_text)
        self.num_output_tokens += len(delta_token_ids)

        if delta is not None:
            delta = maybe_filter_parallel_tool_calls(delta, self.request)
        if finish_reason is not None:
            finish_reason = resolve_finish_reason(
                finish_reason, self.request, self.tools_streamed
            )
        return ChatStreamStep(delta, finish_reason)

step(delta_text, delta_token_ids, finish_reason, prompt_token_ids)

Parse one chunk and resolve its chat finish_reason.

Parameters:

  • delta_text

    (str) –

    Text decoded from this chunk's tokens.

  • delta_token_ids

    (list[int]) –

    This chunk's token IDs.

  • finish_reason

    (str | None) –

    The engine finish reason. None if not finished.

  • prompt_token_ids

    (list[int] | None) –

    Prompt token IDs if known.

Returns:

Source code in vllm/entrypoints/openai/chat_completion/stream_processor.py 
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def step(
    self,
    delta_text: str,
    delta_token_ids: list[int],
    finish_reason: str | None,
    prompt_token_ids: list[int] | None,
) -> ChatStreamStep:
    """Parse one chunk and resolve its chat `finish_reason`.

    Args:
        delta_text: Text decoded from this chunk's tokens.
        delta_token_ids: This chunk's token IDs.
        finish_reason: The engine finish reason. None if not finished.
        prompt_token_ids: Prompt token IDs if known.

    Returns:
        The delta to stream, filtered for `tool_choice` and
        `parallel_tool_calls`, and the chat `finish_reason`.

    """
    delta: DeltaMessage | None
    if not delta_text and not delta_token_ids and not self.num_output_tokens:
        # Finished without output. Send the final chunk unparsed.
        delta = DeltaMessage()
    elif self.parser is not None:
        delta = self.parser.parse_delta(
            delta_text=delta_text,
            delta_token_ids=delta_token_ids,
            request=self.request,
            prompt_token_ids=prompt_token_ids,
            finished=finish_reason is not None,
        )
        if delta is not None and delta.tool_calls:
            if self.keep_tool_calls:
                self.tools_streamed = True
            else:
                delta.tool_calls = []
                if not delta.content and not delta.reasoning:
                    delta = None
    else:
        delta = DeltaMessage(content=delta_text)
    self.num_output_tokens += len(delta_token_ids)

    if delta is not None:
        delta = maybe_filter_parallel_tool_calls(delta, self.request)
    if finish_reason is not None:
        finish_reason = resolve_finish_reason(
            finish_reason, self.request, self.tools_streamed
        )
    return ChatStreamStep(delta, finish_reason)

ChatStreamStep

Bases: NamedTuple

Attributes:

  • delta (DeltaMessage | None) –

    None when the parser has nothing to emit for this chunk yet.

  • finish_reason (str | None) –

    The chat finish_reason or None while the choice is still running.

Source code in vllm/entrypoints/openai/chat_completion/stream_processor.py 
17
18
19
20
21
class ChatStreamStep(NamedTuple):
    delta: DeltaMessage | None
    """None when the parser has nothing to emit for this chunk yet."""
    finish_reason: str | None
    """The chat `finish_reason` or None while the choice is still running."""

delta instance-attribute

None when the parser has nothing to emit for this chunk yet.

finish_reason instance-attribute

The chat finish_reason or None while the choice is still running.