class DelegatingParser(Parser):
"""A Parser implementation that delegates to separate ReasoningParser and
ToolParser instances.
This is the recommended base class for creating model-specific parsers
that combine existing reasoning and tool parser implementations.
Subclasses should set `self._reasoning_parser` and `self._tool_parser`
in their `__init__` method.
If either parser is None, the corresponding methods will return default
values (no reasoning extraction, no tool calls).
"""
def extract_reasoning(
self,
model_output: str,
request: ChatCompletionRequest | ResponsesRequest,
) -> tuple[str | None, str | None]:
if self._reasoning_parser is None:
return None, model_output
return self._reasoning_parser.extract_reasoning(model_output, request)
def _get_function_name(
self, request: ChatCompletionRequest | ResponsesRequest
) -> str:
if request.tool_choice and isinstance(request.tool_choice, ToolChoiceFunction):
return request.tool_choice.name
if request.tool_choice and isinstance(
request.tool_choice, ChatCompletionNamedToolChoiceParam
):
return request.tool_choice.function.name
raise ValueError("Invalid tool_choice for function name extraction.")
def _make_tool_call_id(self, function_name: str) -> str | None:
state = self._stream_state
if state.tool_call_id_type != "kimi_k2":
return None
tool_call_id = make_tool_call_id(
id_type=state.tool_call_id_type,
func_name=function_name,
idx=state.history_tool_call_cnt,
)
state.history_tool_call_cnt += 1
return tool_call_id
def _extract_tool_calls(
self,
content: str | None,
request: ChatCompletionRequest | ResponsesRequest,
enable_auto_tools: bool = False,
) -> tuple[list[FunctionCall] | None, str | None]:
tool_parser = self._tool_parser
if tool_parser is None:
return [], content
if request.tool_choice == "none":
if self._engine_based:
result = self.extract_tool_calls(content or "", request=request)
return [], result.content
return [], content
supports_required_and_named = tool_parser.supports_required_and_named
is_named_tool_choice = request.tool_choice and isinstance(
request.tool_choice,
(ToolChoiceFunction, ChatCompletionNamedToolChoiceParam),
)
is_required_tool_choice = request.tool_choice == "required"
is_auto_tool_choice = enable_auto_tools and (
request.tool_choice == "auto"
or request.tool_choice is None
or (
not supports_required_and_named
and (is_named_tool_choice or is_required_tool_choice)
)
)
tool_calls = list[FunctionCall]()
if is_named_tool_choice and supports_required_and_named:
if content is None or (isinstance(content, str) and not content.strip()):
return [], None
function_name = self._get_function_name(request)
tool_calls.append(
FunctionCall(
id=self._make_tool_call_id(function_name),
name=function_name,
arguments=content,
)
)
content = None
elif is_required_tool_choice and supports_required_and_named:
# "required" with standard JSON-based parsing
parsed_calls = []
with contextlib.suppress(ValidationError):
content = content or ""
parsed_calls = TypeAdapter(list[FunctionDefinition]).validate_json(
content
)
for tc in parsed_calls:
tool_calls.append(
FunctionCall(
id=self._make_tool_call_id(tc.name),
name=tc.name,
arguments=json.dumps(tc.parameters, ensure_ascii=False),
)
)
content = None
elif is_auto_tool_choice:
# Automatic Tool Call Parsing (also used as fallback for
# required/named when supports_required_and_named=False)
tool_call_info = self.extract_tool_calls(
content if content is not None else "",
request=request,
)
if tool_call_info is not None and tool_call_info.tools_called:
tool_calls.extend(
FunctionCall(
id=tc.id,
name=tc.function.name,
arguments=tc.function.arguments,
)
for tc in tool_call_info.tool_calls
)
content = tool_call_info.content
if content and content.strip() == "":
content = None
else:
# No tool calls.
# For required/named tool choice (when falling back to auto
# parsing), if content is empty or whitespace-only, return
# empty list with None content.
if (is_required_tool_choice or is_named_tool_choice) and (
content is None
or (isinstance(content, str) and not content.strip())
):
return [], None
# No complete tool calls: for engine-based parsers, return
# the tool parser's content, which drops incomplete
# tool-call markup (e.g. a <tool_call> opener truncated by
# max_tokens or a stop string), so the non-streaming path
# matches streaming. Legacy parsers keep their existing
# behavior of returning the raw content.
if self._engine_based and tool_call_info is not None:
return None, tool_call_info.content or None
return None, content
return tool_calls, content
def adjust_request(
self, request: ChatCompletionRequest | ResponsesRequest
) -> ChatCompletionRequest | ResponsesRequest:
if self._reasoning_parser is not None:
request = self._reasoning_parser.adjust_request(request)
if self._tool_parser is not None:
request = self._apply_structural_tag(request)
if self._tool_parser is not None:
request = self._tool_parser.adjust_request(request)
return request
def _apply_structural_tag(
self, request: ChatCompletionRequest | ResponsesRequest
) -> ChatCompletionRequest | ResponsesRequest:
if (
self._tool_parser is None
or self._tool_parser.structural_tag_model is None
or not request.tools
):
return request
need_tool_calling = (
request.tool_choice == "auto"
or request.tool_choice == "required"
or isinstance(
request.tool_choice,
(ChatCompletionNamedToolChoiceParam, ToolChoiceFunction),
)
)
if not need_tool_calling:
return request
structure_tag = self._tool_parser.get_structural_tag(
request,
reasoning=False,
strict_level=self.tool_strict_level,
)
if structure_tag is None:
return request
structural_tag = json.dumps(structure_tag.model_dump())
request.structured_outputs = StructuredOutputsParams(
structural_tag=structural_tag,
)
if isinstance(request, ResponsesRequest):
request.text = None
else:
request.response_format = None
return request
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:
if self._reasoning_parser is None:
return DeltaMessage(content=delta_text)
return self._reasoning_parser.extract_reasoning_streaming(
previous_text,
current_text,
delta_text,
previous_token_ids,
current_token_ids,
delta_token_ids,
)
def extract_tool_calls(
self,
model_output: str,
request: ChatCompletionRequest | ResponsesRequest,
) -> ExtractedToolCallInformation:
if self._tool_parser is None:
return ExtractedToolCallInformation(
tools_called=False, tool_calls=[], content=model_output
)
result = None
is_tool_called: bool | Exception = False
try:
result = self._tool_parser.extract_tool_calls(
model_output,
request=request, # type: ignore[arg-type]
)
is_tool_called = bool(result.tools_called)
except Exception as e:
is_tool_called = e
raise
finally:
record_tool_parser_invocation(
is_tool_called=is_tool_called,
is_streaming=False,
request=request,
)
return result
def extract_tool_calls_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],
request: ChatCompletionRequest | ResponsesRequest,
) -> DeltaMessage | None:
if self._tool_parser is None:
return None
result = None
is_tool_called: bool | Exception = False
try:
result = self._tool_parser.extract_tool_calls_streaming(
previous_text,
current_text,
delta_text,
previous_token_ids,
current_token_ids,
delta_token_ids,
request, # type: ignore[arg-type]
)
is_tool_called = bool(result and result.tool_calls)
except Exception as e:
is_tool_called = e
raise
finally:
record_tool_parser_invocation(
is_tool_called=is_tool_called,
is_streaming=True,
request=request,
)
return result
def _extract_tool_calls_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],
request: ChatCompletionRequest | ResponsesRequest,
# The following parameters are used for "required" tool choice parsing and are
# tracked in StreamState for streaming parsing.
tool_call_idx: int | None = None,
tool_call_id_type: str = "random",
function_name_returned: bool = False,
) -> tuple[DeltaMessage | None, bool]:
assert self._tool_parser is not None
supports_required_and_named = self._tool_parser.supports_required_and_named
if request.tool_choice == "none":
if self._engine_based:
# Engine-backed parsers route content extraction through
# extract_tool_calls_streaming, so run the full pipeline
# and strip tool_calls after.
delta_message = self.extract_tool_calls_streaming(
previous_text,
current_text,
delta_text,
previous_token_ids,
current_token_ids,
delta_token_ids,
request, # type: ignore[arg-type]
)
if delta_message:
delta_message.tool_calls = []
return delta_message, False
return (DeltaMessage(content=delta_text) if delta_text else None), False
if (
supports_required_and_named
and request.tool_choice
and isinstance(
request.tool_choice,
(ToolChoiceFunction, ChatCompletionNamedToolChoiceParam),
)
):
delta_message, function_name_returned = extract_named_tool_call_streaming(
delta_text=delta_text,
function_name=self._get_function_name(request),
function_name_returned=function_name_returned,
tool_call_idx=tool_call_idx,
tool_call_id_type=tool_call_id_type,
tokenizer=self.model_tokenizer,
)
return delta_message, function_name_returned
if supports_required_and_named and request.tool_choice == "required":
delta_message, function_name_returned = (
extract_required_tool_call_streaming(
previous_text=previous_text,
current_text=current_text,
delta_text=delta_text,
function_name_returned=function_name_returned,
tool_call_idx=tool_call_idx,
tool_call_id_type=tool_call_id_type,
)
)
return delta_message, function_name_returned
return self.extract_tool_calls_streaming(
previous_text,
current_text,
delta_text,
previous_token_ids,
current_token_ids,
delta_token_ids,
request,
), False
def is_reasoning_end(self, input_ids: list[int]) -> bool:
if self._reasoning_parser is None:
return False
return self._reasoning_parser.is_reasoning_end(input_ids)
def _is_reasoning_end_streaming(
self, input_ids: list[int], delta_ids: list[int]
) -> bool:
if self._reasoning_parser is None:
return False
return self._reasoning_parser.is_reasoning_end_streaming(input_ids, delta_ids)
def _extract_content_ids(self, input_ids: list[int]) -> list[int]:
if self._reasoning_parser is None:
return input_ids
return self._reasoning_parser.extract_content_ids(input_ids)
def _in_reasoning_phase(self, state: StreamState) -> bool:
if self._reasoning_parser is None:
return False
return not state.reasoning_ended
def _in_tool_call_phase(self, state: StreamState) -> bool:
if self._tool_parser is None:
return False
return state.reasoning_ended
def _append_unstreamed_tool_args(
self,
delta_message: DeltaMessage | None,
) -> None:
"""Append parsed-but-unstreamed tool-call arguments to *delta_message*."""
if (
self._tool_parser is not None
and delta_message
and delta_message.tool_calls
and (last_tc := delta_message.tool_calls[-1]).function
):
last_tc.function.arguments = (
last_tc.function.arguments or ""
) + self._tool_parser.get_remaining_unstreamed_args()
def finalize_generation(
self,
delta_message: DeltaMessage | None,
request: ChatCompletionRequest | ResponsesRequest,
state: StreamState,
) -> DeltaMessage | None:
"""Finalize generation for cases where generation was incomplete.
For example, if streaming terminated before reasoning ended
"""
fallback_fn = getattr(
self._reasoning_parser, "get_streaming_fallback_content", None
)
if fallback_fn is not None and not state.reasoning_ended:
promoted = fallback_fn(state.previous_text, request)
if promoted:
if delta_message is None:
delta_message = DeltaMessage()
delta_message.content = (delta_message.content or "") + promoted
self._append_unstreamed_tool_args(delta_message)
return delta_message
def parse(
self,
model_output: str,
request: ChatCompletionRequest | ResponsesRequest,
enable_auto_tools: bool = False,
model_output_token_ids: Sequence[int] = (),
) -> tuple[str | None, str | None, list[FunctionCall] | None]:
self._initialize_history_tool_call_cnt(request)
reasoning, content = self.extract_reasoning(model_output, request)
tool_calls, content = self._extract_tool_calls(
content=content,
request=request,
enable_auto_tools=enable_auto_tools,
)
return reasoning, content, tool_calls
def parse_delta(
self,
delta_text: str,
delta_token_ids: list[int],
request: ChatCompletionRequest | ResponsesRequest,
prompt_token_ids: list[int] | None = None,
*,
finished: bool,
) -> DeltaMessage | None:
self._initialize_history_tool_call_cnt(request)
state = self._stream_state
if not state.prompt_reasoning_checked and prompt_token_ids is not None:
state.prompt_reasoning_checked = True
if self._reasoning_parser is None or self.is_reasoning_end(
prompt_token_ids
):
state.reasoning_ended = True
else:
# Reasoning is still open at the end of the prompt; let the
# reasoning parser adjust its initial parsing state so the
# first generated tokens are classified correctly.
self._reasoning_parser.adjust_initial_state_from_prompt(
prompt_token_ids
)
current_text, current_token_ids = state.advance(delta_text, delta_token_ids)
delta_message: DeltaMessage | None = None
reasoning_transitioned = False
# Reasoning extraction
if self._in_reasoning_phase(state):
delta_message = self.extract_reasoning_streaming(
previous_text=state.previous_text,
current_text=current_text,
delta_text=delta_text,
previous_token_ids=state.previous_token_ids,
current_token_ids=current_token_ids,
delta_token_ids=delta_token_ids,
)
reasoning_parser = self._reasoning_parser
if reasoning_parser is not None and reasoning_parser.engine_based_streaming:
should_transition = (
reasoning_parser.has_engine_confirmed_reasoning_end()
)
else:
should_transition = self._is_reasoning_end_streaming(
current_token_ids, delta_token_ids
)
if should_transition:
state.reasoning_ended = True
reasoning_transitioned = True
current_token_ids = self._extract_content_ids(delta_token_ids)
# Flush whenever the reasoning parser is engine-based (not only
# when _engine_based is True): it buffers the post-marker text
# (e.g. the "<" of "<tool_call>"), surfaced via finish_streaming().
flush_delta = (
reasoning_parser.finish_streaming() # type: ignore[union-attr, attr-defined]
if reasoning_parser is not None
and reasoning_parser.engine_based_streaming
else None
)
current_text = (
(delta_message.content if delta_message else None) or ""
) + ((flush_delta.content if flush_delta else None) or "")
if self._engine_based:
if delta_message and self._tool_parser is not None:
delta_message.content = None
else:
delta_text = current_text
# Tool call extraction
if self._in_tool_call_phase(state):
if not state.tool_call_text_started:
state.tool_call_text_started = True
state.previous_text = ""
state.previous_token_ids = []
delta_text = current_text
delta_token_ids = current_token_ids
reasoning_from_this_batch = (
delta_message.reasoning if delta_message else None
)
delta_message, state.function_name_returned = (
self._extract_tool_calls_streaming(
previous_text=state.previous_text,
current_text=current_text,
delta_text=delta_text,
previous_token_ids=state.previous_token_ids,
current_token_ids=current_token_ids,
delta_token_ids=delta_token_ids,
request=request, # type: ignore[arg-type]
tool_call_idx=state.history_tool_call_cnt,
tool_call_id_type=state.tool_call_id_type,
function_name_returned=state.function_name_returned,
)
)
if reasoning_from_this_batch:
if delta_message is None:
delta_message = DeltaMessage(reasoning=reasoning_from_this_batch)
elif not delta_message.reasoning:
delta_message.reasoning = reasoning_from_this_batch
if (
delta_message
and delta_message.tool_calls
and delta_message.tool_calls[0].id is not None
):
state.history_tool_call_cnt += 1
# No phase active: pass through as content.
# Skip when reasoning just ended in this delta — the engine already
# consumed the end-of-reasoning marker (e.g. </think>) and
# delta_text still contains the raw marker text.
if (
delta_message is None
and not reasoning_transitioned
and not self._in_reasoning_phase(state)
and not self._in_tool_call_phase(state)
):
delta_message = DeltaMessage(content=delta_text)
state.commit(current_text, current_token_ids)
if finished:
delta_message = self.finalize_generation(delta_message, request, state)
delta_message = self._flush_engine_parsers(delta_message)
# Suppress reasoning deltas if not requested
if delta_message and not request.include_reasoning:
delta_message.reasoning = None
# If only reasoning was in the message (no content, no tool_calls)
# skip emitting entirely
if not delta_message.content and not delta_message.tool_calls:
delta_message = None
return delta_message
def count_reasoning_tokens(self, token_ids: Sequence[int]) -> int:
"""Count reasoning tokens through the configured reasoning parser."""
if self._reasoning_parser is None:
return 0
return self._reasoning_parser.count_reasoning_tokens(token_ids)
def _flush_engine_parsers(
self, delta_message: DeltaMessage | None
) -> DeltaMessage | None:
"""Flush buffered state from engine-based parsers at stream end."""
reasoning_ended = self._stream_state.reasoning_ended
for parser in (self._reasoning_parser, self._tool_parser):
if not getattr(parser, "engine_based_streaming", False):
continue
# When reasoning has ended and we transitioned to the tool
# phase, the reasoning parser's engine may still have buffered
# characters from tool-call markup it saw with
# skip_tool_parsing=True. Flushing that would leak spurious
# content (e.g. a stray '"'), so skip it.
if parser is self._reasoning_parser and reasoning_ended:
continue
finish = getattr(parser, "finish_streaming", None)
if finish is None:
continue
flush_delta = finish()
if flush_delta is None:
continue
if delta_message is None:
delta_message = flush_delta
else:
if flush_delta.content:
delta_message.content = (
delta_message.content or ""
) + flush_delta.content
if flush_delta.reasoning:
delta_message.reasoning = (
delta_message.reasoning or ""
) + flush_delta.reasoning
if flush_delta.tool_calls:
delta_message.tool_calls = (
delta_message.tool_calls or []
) + flush_delta.tool_calls
return delta_message