class ResponseTemplateReasoningParser(ReasoningParser):
"""Reasoning end detection for `--reasoning-parser hf`.
`ResponseTemplateParser` parses the output; this parser tells it and the
structured-output engine where reasoning ends.
"""
def __init__(
self,
tokenizer: TokenizerLike,
*args,
response_template: dict[str, Any] | None = None,
**kwargs,
) -> None:
super().__init__(tokenizer, *args, **kwargs)
self.response_template = _serving_template(tokenizer, response_template)
chat_template_kwargs = kwargs.get("chat_template_kwargs") or {}
self._thinking_disabled = not chat_template_kwargs.get("enable_thinking", True)
self._reasoning_end_ids = _reasoning_end_ids(
tokenizer, self.vocab, self.response_template
)
@property
def reasoning_start_str(self) -> str | None:
thinking = self.response_template.fields.get(THINKING_FIELD)
literals = thinking.open_literals if thinking else None
return literals[0] if literals else None
@property
def reasoning_end_str(self) -> str | None:
thinking = self.response_template.fields.get(THINKING_FIELD)
literals = thinking.close_literals if thinking else None
return literals[0] if literals else None
def is_reasoning_end(self, input_ids: Sequence[int]) -> bool:
try:
parser = ResponseParser(
self.response_template,
prefix=_decode(self.model_tokenizer, input_ids),
)
except Exception:
return False
return _reasoning_has_ended(
parser,
self.response_template,
thinking_disabled=self._thinking_disabled,
)
def is_reasoning_end_streaming(
self, input_ids: Sequence[int], delta_ids: Iterable[int]
) -> bool:
if self._reasoning_end_ids is None:
return self.is_reasoning_end(input_ids)
num_new = len(list(delta_ids))
return any(
_ends_within(input_ids, end_ids, num_new)
for end_ids in self._reasoning_end_ids
)
def extract_content_ids(self, input_ids: list[int]) -> list[int]:
raise NotImplementedError("ResponseTemplateParser extracts content")
def extract_reasoning(
self,
model_output: str,
request: ChatCompletionRequest | ResponsesRequest,
) -> tuple[str | None, str | None]:
raise NotImplementedError("ResponseTemplateParser extracts reasoning")
def extract_reasoning_streaming(
self,
previous_text: str,
current_text: str,
delta_text: str,
previous_token_ids: Sequence[int],
current_token_ids: Sequence[int],
delta_token_ids: Sequence[int],
) -> DeltaMessage | None:
raise NotImplementedError("ResponseTemplateParser extracts reasoning")