Skip to content

vllm.parser.granite_thinking

Granite 4.2 thinking parser.

Granite 4.2 uses the same tool call and reasoning format as Nemotron V3 (<think>/</think> + <tool_call> XML). This config reuses :class:NemotronV3Parser with two Granite-specific behaviors:

  1. Strip leading newlines from content — the Granite chat template writes \n</think>\n, so the first content character is always \n.
  2. Inherit the Nemotron V3 reasoning-to-content swap for enable_thinking=False / force_nonempty_content=True.

Classes:

GraniteThinkingParser

Bases: NemotronV3Parser

Granite 4.2 parser: same format as Nemotron V3, with leading-newline stripping on content after </think>.

The reasoning_ended property is overridden to delay the reasoning-to-content transition reported to :class:~vllm.parser.abstract_parser.DelegatingParser until the template's leading newline has been consumed. Without this delay, DelegatingParser.parse_delta passes newline tokens through a raw pass-through path that bypasses the engine's _events_to_delta (and therefore our stripping logic).

Source code in vllm/parser/granite_thinking.py 
 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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
class GraniteThinkingParser(NemotronV3Parser):
    """Granite 4.2 parser: same format as Nemotron V3, with
    leading-newline stripping on content after ``</think>``.

    The ``reasoning_ended`` property is overridden to delay the
    reasoning-to-content transition reported to
    :class:`~vllm.parser.abstract_parser.DelegatingParser` until the
    template's leading newline has been consumed.  Without this delay,
    ``DelegatingParser.parse_delta`` passes newline tokens through a
    raw pass-through path that bypasses the engine's
    ``_events_to_delta`` (and therefore our stripping logic).
    """

    def __init__(
        self,
        tokenizer: TokenizerLike,
        tools: list[Tool] | None = None,
        **kwargs,
    ) -> None:
        chat_kwargs = kwargs.get("chat_template_kwargs", {}) or {}
        thinking = chat_kwargs.get("enable_thinking", True)
        kwargs.pop("parser_engine_config", None)
        Qwen3Parser.__init__(
            self,
            tokenizer,
            tools,
            parser_engine_config=granite_thinking_config(thinking=thinking),
            **kwargs,
        )
        self._streamed_reasoning: list[str] = []
        self._content_started = False

    @property
    def reasoning_ended(self) -> bool:
        # Internal engine logic uses _reasoning_ended directly; this
        # property is only read by DelegatingParser (via the adapter's
        # has_engine_confirmed_reasoning_end) to decide when to stop
        # routing deltas through extract_reasoning_streaming.
        if not self._reasoning_ended:
            return False
        return self._content_started

    def _reset(self, initial_state=None) -> None:
        super()._reset(initial_state=initial_state)
        self._content_started = False

    def _events_to_delta(
        self,
        events: list[SemanticEvent],
        finished: bool = False,
    ) -> DeltaMessage | None:
        delta = super()._events_to_delta(events, finished=finished)
        if (
            delta is not None
            and delta.content is not None
            and not self._content_started
        ):
            stripped = delta.content.lstrip("\n")
            if not stripped:
                delta.content = None
                if delta.reasoning is None and not delta.tool_calls:
                    return None
            else:
                self._content_started = True
                delta.content = stripped
        return delta

    def extract_reasoning(
        self,
        model_output: str,
        request: ChatCompletionRequest | ResponsesRequest,
    ) -> tuple[str | None, str | None]:
        reasoning, content = Qwen3Parser.extract_reasoning(self, model_output, request)

        # Track whether content was truly absent (None) vs present
        # but whitespace-only — the swap should only fire for absent.
        content_was_absent = content is None

        if content is not None:
            content = content.lstrip("\n") or None

        # Only swap when content was truly absent from model output
        # (truncated reasoning or max_tokens cut at </think>), not
        # when it was present but stripped to empty (newlines-only).
        if self._should_force_content(request) and content_was_absent:
            reasoning, content = content, reasoning

        return reasoning, content