from __future__ import annotations import copy import json import logging import time import uuid from enum import Enum from http import HTTPStatus from typing import TYPE_CHECKING, Any, AsyncGenerator, Dict, List, Optional, Union class ThinkingMode(str, Enum): """Mode for message encoding - chat vs thinking/reasoning.""" CHAT = "chat" THINKING = "thinking" import jinja2 import orjson from fastapi import Request from fastapi.responses import ORJSONResponse, StreamingResponse from jsonschema import Draft202012Validator, SchemaError from sglang.srt.entrypoints.openai import encoding_dsv4, encoding_dsv32 from sglang.srt.entrypoints.openai.protocol import ( ChatCompletionRequest, ChatCompletionResponse, ChatCompletionResponseChoice, ChatCompletionResponseStreamChoice, ChatCompletionStreamResponse, ChatCompletionTokenLogprob, ChatMessage, ChoiceLogprobs, DeltaMessage, ErrorResponse, FunctionResponse, LogProbs, MessageProcessingResult, PromptTokensDetails, ResponseParserProtocol, SglExt, ToolCall, ToolCallProcessingResult, ToolChoice, TopLogprob, ) from sglang.srt.entrypoints.openai.serving_base import OpenAIServingBase from sglang.srt.entrypoints.openai.sse_utils import build_sse_content from sglang.srt.entrypoints.openai.usage_processor import UsageProcessor from sglang.srt.entrypoints.openai.utils import ( cached_tokens_details_from_dict, process_cached_tokens_details_from_ret, process_hidden_states_from_ret, process_routed_experts_from_ret, should_include_usage, to_openai_style_logprobs, ) from sglang.srt.environ import envs from sglang.srt.function_call.core_types import ToolCallItem from sglang.srt.function_call.function_call_parser import FunctionCallParser from sglang.srt.function_call.json_array_parser import JsonArrayParser from sglang.srt.function_call.utils import ( get_json_schema_constraint, normalize_json_schema_types, ) from sglang.srt.managers.io_struct import GenerateReqInput from sglang.srt.parser.conversation import generate_chat_conv from sglang.srt.parser.jinja_template_utils import process_content_for_template_format from sglang.srt.parser.reasoning_parser import ReasoningParser if TYPE_CHECKING: from sglang.srt.managers.tokenizer_manager import TokenizerManager from sglang.srt.parser.template_manager import TemplateManager logger = logging.getLogger(__name__) def normalize_tool_content(role: str, content): """Normalize tool message content from OpenAI array format to plain string. OpenAI clients may send tool content as a list of content parts (e.g. [{"type":"text","text":"..."}]) but most chat templates expect a plain string for tool messages. Only flatten when ALL items are pure OpenAI text parts; preserve lists containing non-text-type items that some templates intentionally iterate over. """ if role != "tool" or not isinstance(content, list): return content parts = content is_openai_text_parts = all( (isinstance(p, dict) and p.get("type") == "text") or isinstance(p, str) for p in parts ) if is_openai_text_parts: text_parts = [p.get("text", "") if isinstance(p, dict) else p for p in parts] return " ".join(text_parts) return content def parse_tool_call_arguments(arguments: str) -> Dict[str, Any]: """Parse OpenAI tool call arguments for chat templates.""" try: parsed_arguments = orjson.loads(arguments) except orjson.JSONDecodeError as exc: raise ValueError( "Assistant tool call function.arguments must be valid JSON." ) from exc if not isinstance(parsed_arguments, dict): raise ValueError( "Assistant tool call function.arguments must be a JSON object." ) return parsed_arguments def normalize_assistant_tool_call_arguments(message: Dict[str, Any]) -> None: """Normalize assistant history tool call arguments in-place.""" if message.get("role") != "assistant" or not isinstance( message.get("tool_calls"), list ): return for item in message["tool_calls"]: function = item.get("function") if isinstance(item, dict) else None if not isinstance(function, dict): continue if "arguments" in function and isinstance(function["arguments"], str): function["arguments"] = parse_tool_call_arguments(function["arguments"]) def _extract_max_dynamic_patch(request: ChatCompletionRequest): img_vals = [] vid_vals = [] for msg in request.messages or []: content = getattr(msg, "content", None) if not isinstance(content, list): continue for part in content: # pydantic object or dict type if getattr(part, "type", None) == "image_url": iu = getattr(part, "image_url", None) mdp = getattr(iu, "max_dynamic_patch", None) if iu else None if mdp is not None: img_vals.append(int(mdp)) elif getattr(part, "type", None) == "video_url": vu = getattr(part, "video_url", None) mdp = getattr(vu, "max_dynamic_patch", None) if vu else None if mdp is not None: vid_vals.append(int(mdp)) # TODO(yuan-luo): per-item max_dynamic_patch for both image and video img_max_dynamic_patch = min(img_vals) if img_vals else None vid_max_dynamic_patch = min(vid_vals) if vid_vals else None return img_max_dynamic_patch, vid_max_dynamic_patch class OpenAIServingChat(OpenAIServingBase): """Handler for /v1/chat/completions requests""" _default_sampling_params_logged = False def __init__( self, tokenizer_manager: TokenizerManager, template_manager: TemplateManager, ): super().__init__(tokenizer_manager) self.template_manager = template_manager self.tool_call_parser = self.tokenizer_manager.server_args.tool_call_parser self.reasoning_parser = self.tokenizer_manager.server_args.reasoning_parser self._reasoning_detector = None if self.reasoning_parser: try: rp = ReasoningParser( model_type=self.reasoning_parser, stream_reasoning=True, tokenizer=self.tokenizer_manager.tokenizer, ) self._reasoning_detector = rp.detector except ValueError as e: logger.warning( "Failed to initialize reasoning detector for parser '%s': %s", self.reasoning_parser, e, ) # Get default sampling parameters from model's generation config self.default_sampling_params = ( self.tokenizer_manager.model_config.get_default_sampling_params() ) if ( self.default_sampling_params and not OpenAIServingChat._default_sampling_params_logged ): logger.info( f"Using default chat sampling params from model generation config: {self.default_sampling_params}", ) OpenAIServingChat._default_sampling_params_logged = True # Check if the model is a GPT-OSS model self.is_gpt_oss = ( hasattr(self.tokenizer_manager.model_config, "hf_config") and hasattr(self.tokenizer_manager.model_config.hf_config, "model_type") and self.tokenizer_manager.model_config.hf_config.model_type == "gpt_oss" ) self.is_gemma4 = ( hasattr(self.tokenizer_manager.model_config, "hf_config") and hasattr(self.tokenizer_manager.model_config.hf_config, "model_type") and self.tokenizer_manager.model_config.hf_config.model_type in ("gemma4", "gemma4_unified") ) # Which Python-based chat encoder (if any) bypasses apply_chat_template. # Values: "dsv32", "dsv4", or custom values set by subclass. None for default. self.chat_encoding_spec = self._resolve_chat_encoding_spec() # Per-request response parser for custom decoding (set by _encode_messages) self._response_parser: Optional[ResponseParserProtocol] = None # Probe whether ``encode("")`` returns specials. If it does, we must # keep ``add_special_tokens=False`` at the chat-template encode site # to avoid double BOS; otherwise the kwarg is a no-op and dropping it # lets slow tokenizers (e.g. Kimi's TikTokenTokenizer) stay on the # fast internal path. try: self._tokenizer_auto_adds_specials = ( len(self.tokenizer_manager.tokenizer.encode("")) > 0 ) except Exception: self._tokenizer_auto_adds_specials = True def _handle_last_assistant_message( self, messages: List[Dict[str, Any]], request: ChatCompletionRequest, ) -> tuple[List[Dict[str, Any]], Optional[str]]: """ Handle continue_final_message feature: separate final assistant message. If continue_final_message is enabled and the last message is from assistant, extract its content and remove it from the message list. If continue_final_message is False and the last message is from assistant, convert it to a user message to ensure the last message is always from user. Only processes text-based content (strings), ignoring multimodal content (lists). Args: messages: List of message dictionaries request: ChatCompletionRequest with continue_final_message flag Returns: Tuple of (processed_messages, assistant_prefix) - processed_messages: Messages with last assistant message handled appropriately - assistant_prefix: Content of the last assistant message (string only), or None """ assistant_prefix = None if messages and messages[-1].get("role") == "assistant": last_content = messages[-1].get("content") # Only process string content, ignore multimodal content (lists) if isinstance(last_content, str): if request.continue_final_message: # Extract content and remove the assistant message assistant_prefix = last_content messages = messages[:-1] else: # Convert the last assistant message to user message messages[-1] = {"role": "user", "content": last_content} return messages, assistant_prefix def _append_assistant_prefix_to_prompt_ids( self, prompt_ids: List[int], assistant_prefix: str ) -> List[int]: """ Append assistant prefix to prompt_ids. Args: prompt_ids: Current prompt token IDs assistant_prefix: Assistant message content to append Returns: Updated prompt_ids with assistant prefix appended """ encoded = self.tokenizer_manager.tokenizer.encode(assistant_prefix) if encoded and encoded[0] == self.tokenizer_manager.tokenizer.bos_token_id: encoded = encoded[1:] return prompt_ids + encoded def _resolve_chat_encoding_spec(self) -> Optional[str]: """Determine which chat encoding spec to use. Override in subclass to add custom encoding specs. """ from sglang.srt.entrypoints.openai.chat_encoding import ( resolve_chat_encoding_spec, ) return resolve_chat_encoding_spec( hf_config=self.tokenizer_manager.model_config.hf_config, tokenizer=self.tokenizer_manager.tokenizer, tool_call_parser=self.tool_call_parser, ) def _request_id_prefix(self) -> str: return "chatcmpl-" def _encode_messages( self, messages: List[Dict[str, Any]], request: ChatCompletionRequest, thinking_mode: ThinkingMode, ) -> Optional[List[int]]: """Encode messages for custom chat_encoding_spec values. Returns prompt_ids if handled, None to use default encoding. """ return None def _decode_response(self, ret_item: Dict[str, Any]) -> Union[str, ErrorResponse]: """Extract text from response.""" return ret_item["text"] def _get_parsed_response_fields( self, reasoning_text: Optional[str], tool_calls: Optional[List[Dict]], ) -> tuple[Optional[str], Optional[List[Dict]]]: """Post-process reasoning and tool_calls before building response.""" return reasoning_text, tool_calls def _continuous_usage_cached_details( self, content: Dict[str, Any] ) -> Optional[PromptTokensDetails]: if not self.tokenizer_manager.server_args.enable_cache_report: return None return UsageProcessor._details_if_cached( content["meta_info"].get("cached_tokens", 0) ) async def _generate_stream_content( self, content: Dict[str, Any], index: int, request: ChatCompletionRequest, stream_offsets: Dict[int, int], reasoning_parser_dict: Dict, parser_dict: Dict, has_tool_calls: Dict[int, bool], choice_logprobs: Optional[Dict], finish_reason_type: Optional[str], continuous_usage_stats: bool, prompt_tokens: Dict[int, int], reasoning_tokens: Dict[int, int], completion_tokens: Dict[int, int], ) -> AsyncGenerator[str, None]: """Generate SSE chunks for streaming content.""" offset = stream_offsets.get(index, 0) if self.tokenizer_manager.server_args.incremental_streaming_output: delta = content["text"] else: delta = content["text"][offset:] stream_offsets[index] = len(content["text"]) # Attach logprobs to the first chunk emitted this step (reasoning, # tool-call, or content) so they aren't dropped when a parser is active # nor duplicated across chunks; flush any leftover at the end. remaining_logprobs = choice_logprobs # Handle reasoning content if self.reasoning_parser and request.separate_reasoning: reasoning_text, delta = self._process_reasoning_stream( index, delta, reasoning_parser_dict, content, request ) if reasoning_text: usage = None if continuous_usage_stats: usage = UsageProcessor.calculate_token_usage( prompt_tokens=prompt_tokens.get(index, 0), reasoning_tokens=reasoning_tokens.get(index, 0), completion_tokens=completion_tokens.get(index, 0), cached_tokens=self._continuous_usage_cached_details(content), ).model_dump() yield build_sse_content( chunk_id=content["meta_info"]["id"], created=int(time.time()), model=request.model, index=index, reasoning_content=reasoning_text, logprobs=remaining_logprobs, usage=usage, ) remaining_logprobs = None # Handle tool calls if request.tool_choice != "none" and request.tools and self.tool_call_parser: async for chunk in self._process_tool_call_stream( index, delta, parser_dict, content, request, has_tool_calls, continuous_usage_stats, ): if chunk: yield chunk # Send any remaining tool call arguments when generation finishes if finish_reason_type is not None and index in parser_dict: parser = parser_dict[index] remaining_chunk = self._check_for_unstreamed_tool_args( parser, content, request, index ) if remaining_chunk: yield remaining_chunk else: # Regular content if delta: usage = None if continuous_usage_stats: usage = UsageProcessor.calculate_token_usage( prompt_tokens=prompt_tokens.get(index, 0), reasoning_tokens=reasoning_tokens.get(index, 0), completion_tokens=completion_tokens.get(index, 0), cached_tokens=self._continuous_usage_cached_details(content), ).model_dump() yield build_sse_content( chunk_id=content["meta_info"]["id"], created=int(time.time()), model=request.model, index=index, content=delta, logprobs=remaining_logprobs, usage=usage, ) remaining_logprobs = None # Flush logprobs still unattached this step — only when a parser is # active, since _process_tool_call_stream may consume the delta and emit # no content chunk. On the plain path an empty-delta step has no chunk # to attach to either way, and a standalone empty-delta logprobs chunk # is not a shape clients expect. if remaining_logprobs is not None and ( self.reasoning_parser or self.tool_call_parser ): usage = None if continuous_usage_stats: usage = UsageProcessor.calculate_token_usage( prompt_tokens=prompt_tokens.get(index, 0), reasoning_tokens=reasoning_tokens.get(index, 0), completion_tokens=completion_tokens.get(index, 0), cached_tokens=self._continuous_usage_cached_details(content), ).model_dump() yield build_sse_content( chunk_id=content["meta_info"]["id"], created=int(time.time()), model=request.model, index=index, logprobs=remaining_logprobs, usage=usage, ) def _validate_request(self, request: ChatCompletionRequest) -> Optional[str]: """Validate that the input is valid.""" if not request.messages: return "Messages cannot be empty." if ( isinstance(request.tool_choice, str) and request.tool_choice.lower() == "required" and not request.tools ): return "Tools cannot be empty if tool choice is set to required." if request.tool_choice is not None and not isinstance(request.tool_choice, str): if not request.tools: return "Tools cannot be empty if tool choice is set to a specific tool." tool_name = request.tool_choice.function.name tool_exists = any(tool.function.name == tool_name for tool in request.tools) if not tool_exists: return f"Tool '{tool_name}' not found in tools list." # Validate tool definitions for i, tool in enumerate(request.tools or []): if tool.function.parameters is None: continue try: # Rewrite DB/ORM-style aliases (e.g. "varchar", "enum", "int") # to standard JSON Schema types before validation. RecursionError # guards against hand-crafted cyclic schemas so the request gets # a 400 instead of crashing into a 500. normalize_json_schema_types(tool.function.parameters) Draft202012Validator.check_schema(tool.function.parameters) except SchemaError as e: return f"Tool {i} function has invalid 'parameters' schema: {str(e)}" except RecursionError: return ( f"Tool {i} function 'parameters' schema is too deeply nested " "or contains a cycle." ) max_output_tokens = request.max_completion_tokens or request.max_tokens server_context_length = self.tokenizer_manager.server_args.context_length if ( max_output_tokens and server_context_length and max_output_tokens > server_context_length ) and not self.tokenizer_manager.server_args.allow_auto_truncate: return ( f"max_completion_tokens is too large: {max_output_tokens}." f"This model supports at most {server_context_length} completion tokens." ) if request.response_format and request.response_format.type == "json_schema": schema = getattr(request.response_format.json_schema, "schema_", None) if schema is None: return "schema_ is required for json_schema response format request." return None def _convert_to_internal_request( self, request: ChatCompletionRequest, raw_request: Request = None, ) -> tuple[GenerateReqInput, ChatCompletionRequest]: reasoning_effort = ( request.chat_template_kwargs.pop("reasoning_effort", None) if request.chat_template_kwargs else None ) if self.is_gpt_oss and reasoning_effort == "none": raise ValueError( f"Harmony does not support reasoning effort {reasoning_effort}" ) if reasoning_effort is not None: request.reasoning_effort = reasoning_effort if request.stream: if request.return_prompt_token_ids: raise ValueError( "return_prompt_token_ids is not supported with streaming. " "Please set stream=false when using return_prompt_token_ids=true." ) if request.return_meta_info: raise ValueError( "return_meta_info is not supported with streaming. " "Please set stream=false when using return_meta_info=true." ) is_multimodal = self.tokenizer_manager.model_config.is_multimodal # Process messages and apply chat template processed_messages = self._process_messages(request, is_multimodal) # Build sampling parameters sampling_params = request.to_sampling_params( stop=processed_messages.stop, model_generation_config=self.default_sampling_params, tool_call_constraint=processed_messages.tool_call_constraint, ) if request.input_ids is not None: prompt_kwargs = {"input_ids": processed_messages.prompt_ids} elif is_multimodal: prompt_kwargs = {"text": processed_messages.prompt} else: if isinstance(processed_messages.prompt_ids, str): prompt_kwargs = {"text": processed_messages.prompt_ids} else: prompt_kwargs = {"input_ids": processed_messages.prompt_ids} # Extract custom labels from raw request headers custom_labels = self.extract_custom_labels(raw_request) # Extract routed_dp_rank from header (has higher priority than body) effective_routed_dp_rank = self.extract_routed_dp_rank_from_header( raw_request, request.routed_dp_rank ) # Resolve LoRA adapter from model parameter or explicit lora_path lora_path = self._resolve_lora_path(request.model, request.lora_path) img_max_dynamic_patch, vid_max_dynamic_patch = _extract_max_dynamic_patch( request ) require_reasoning = self._get_reasoning_from_request(request) adapted_request = GenerateReqInput( **prompt_kwargs, image_data=processed_messages.image_data, video_data=processed_messages.video_data, audio_data=processed_messages.audio_data, sampling_params=sampling_params, return_logprob=request.logprobs, logprob_start_len=-1, top_logprobs_num=request.top_logprobs or 0, stream=request.stream, return_text_in_logprobs=True, modalities=processed_messages.modalities, lora_path=lora_path, bootstrap_host=request.bootstrap_host, bootstrap_port=request.bootstrap_port, bootstrap_room=request.bootstrap_room, routed_dp_rank=effective_routed_dp_rank, disagg_prefill_dp_rank=request.disagg_prefill_dp_rank, return_hidden_states=request.return_hidden_states, return_routed_experts=request.return_routed_experts, routed_experts_start_len=request.routed_experts_start_len, rid=request.rid, session_id=request.session_id, extra_key=self._compute_extra_key(request), require_reasoning=require_reasoning, priority=request.priority, routing_key=self.extract_routing_key(raw_request), custom_labels=custom_labels, custom_logit_processor=request.custom_logit_processor, images_config=getattr(request, "images_config", None), image_max_dynamic_patch=img_max_dynamic_patch, video_max_dynamic_patch=vid_max_dynamic_patch, max_dynamic_patch=getattr(request, "max_dynamic_patch", None), use_audio_in_video=getattr(request, "use_audio_in_video", False), return_prompt_token_ids=request.return_prompt_token_ids, ) return adapted_request, request def _process_messages( self, request: ChatCompletionRequest, is_multimodal: bool ) -> MessageProcessingResult: """Process chat messages and apply chat template""" # GptOss model needs to keep special tokens for harmony parsing if self.is_gpt_oss or self.is_gemma4: request.skip_special_tokens = False self._patch_reasoning_skip_special_tokens(request) thinking_mode = self._get_reasoning_from_request(request) # SGLang's ReasonerGrammarBackend owns the reasoning prefix # when --reasoning-parser is configured, so builtin xgrammar # tags must describe only the post-reasoning tool-call suffix. xgrammar_reasoning = thinking_mode and ( self.tokenizer_manager.server_args.reasoning_parser is None ) tool_call_constraint = None # Apply chat template and its stop strings tools = None if request.tools and request.tool_choice != "none": request.skip_special_tokens = False if not isinstance(request.tool_choice, str): tools = [ item.model_dump() for item in request.tools if item.function.name == request.tool_choice.function.name ] else: tools = [item.model_dump() for item in request.tools] if self.tool_call_parser: parser = FunctionCallParser( request.tools, self.tool_call_parser, tokenizer=self.tokenizer_manager.tokenizer, ) tool_call_constraint = parser.get_structure_constraint( request.tool_choice, parallel_tool_calls=request.parallel_tool_calls, thinking_mode=xgrammar_reasoning, ) # Fallback: use generic JSON schema for required/named tool choice # only when no parser-specific constraint was set if tool_call_constraint is None and ( request.tool_choice == "required" or isinstance(request.tool_choice, ToolChoice) ): json_schema = get_json_schema_constraint( request.tools, request.tool_choice, parallel_tool_calls=request.parallel_tool_calls, ) tool_call_constraint = ("json_schema", json_schema) # When input_ids are provided, skip template tokenization entirely; # only stop tokens and tool_call_constraint are needed. if request.input_ids is not None: result = MessageProcessingResult( prompt="", prompt_ids=request.input_ids, image_data=None, audio_data=None, video_data=None, modalities=[], stop=request.stop or [], ) elif self.template_manager.chat_template_name is None: result = self._apply_jinja_template(request, tools, is_multimodal) else: result = self._apply_conversation_template(request, is_multimodal) result.tool_call_constraint = tool_call_constraint return result def _apply_jinja_template( self, request: ChatCompletionRequest, tools: Optional[List[Dict]], is_multimodal: bool, ) -> MessageProcessingResult: """Apply Jinja chat template""" prompt = "" prompt_ids = [] openai_compatible_messages = [] image_data = [] video_data = [] audio_data = [] modalities = [] template_content_format = self.template_manager.jinja_template_content_format # Try custom encoding first (override in subclass for custom renderers) thinking_requested = (request.chat_template_kwargs or {}).get( "thinking", envs.SGLANG_DEFAULT_THINKING.get() ) thinking_mode = ( ThinkingMode.THINKING if thinking_requested else ThinkingMode.CHAT ) messages = [msg.model_dump() for msg in request.messages] for message in messages: normalize_assistant_tool_call_arguments(message) prompt_ids = self._encode_messages( copy.deepcopy(messages), request, thinking_mode ) if prompt_ids is not None: # Custom encoding handled it - no further processing needed pass elif self.chat_encoding_spec is not None: # dsv4/dsv32 encoding path messages = copy.deepcopy(messages) # dsv4/dsv32 are text-only and consume string content; flatten # OpenAI parts-list content here so the encoder sees a plain string. for i, msg in enumerate(messages): if isinstance(msg.get("content"), list): messages[i] = process_content_for_template_format( msg, "string", [], [], [], [] ) for msg in messages: if msg.get("content") is None: msg["content"] = "" processed_msg = process_content_for_template_format( msg, template_content_format, image_data, video_data, audio_data, modalities, use_dpsk_v32_encoding=self.chat_encoding_spec == "dsv32", ) msg.update(processed_msg) # Handle continue_final_message: separate final assistant message messages, assistant_prefix = self._handle_last_assistant_message( messages, request ) if messages[0]["role"] != "system": # insert an empty system prompt to help render tool system prompt messages.insert(0, {"role": "system", "content": ""}) if request.tools: messages[0]["tools"] = [tool.model_dump() for tool in request.tools] # Default encoding (dsv4/dsv32) if self.chat_encoding_spec == "dsv4": # V4 encoder only accepts "max" / "high" / None. # OpenAI protocol defaults to "medium" which V4 rejects; drop it. # Fallback: if request didn't set it, try env SGLANG_DSV4_REASONING_EFFORT. effort_source = request.reasoning_effort if effort_source is None: env_val = envs.SGLANG_DSV4_REASONING_EFFORT.get() if env_val: effort_source = env_val v4_reasoning_effort = ( effort_source if effort_source in ("max", "high") else None ) if request.task is not None: encoding_dsv4.attach_task_to_last_user_message( messages, request.task ) real_input = encoding_dsv4.encode_messages( messages, thinking_mode=thinking_mode, reasoning_effort=v4_reasoning_effort, ) prompt_ids = self.tokenizer_manager.tokenizer.encode(real_input) else: real_input = encoding_dsv32.encode_messages( messages, thinking_mode=thinking_mode ) prompt_ids = self.tokenizer_manager.tokenizer.encode(real_input) # Append assistant prefix if continue_final_message is enabled if assistant_prefix: prompt_ids = self._append_assistant_prefix_to_prompt_ids( prompt_ids, assistant_prefix ) else: for msg_dict in copy.deepcopy(messages): if msg_dict.get("content") is None: msg_dict["content"] = "" # Process content based on detected template format processed_msg = process_content_for_template_format( msg_dict, template_content_format, image_data, video_data, audio_data, modalities, ) processed_msg["content"] = normalize_tool_content( processed_msg["role"], processed_msg.get("content") ) openai_compatible_messages.append(processed_msg) # Handle continue_final_message: separate final assistant message openai_compatible_messages, assistant_prefix = ( self._handle_last_assistant_message(openai_compatible_messages, request) ) extra_template_kwargs = {} if request.reasoning_effort is not None: extra_template_kwargs["reasoning_effort"] = request.reasoning_effort if request.chat_template_kwargs: extra_template_kwargs.update(request.chat_template_kwargs) rc = self.template_manager.reasoning_config if rc is not None and rc.effort_kwarg is not None: if request.reasoning_effort == "low": extra_template_kwargs.setdefault(rc.effort_kwarg, True) elif request.reasoning_effort in ("medium", "high", "max"): logger.warning( "Model '%s' supports only 'low' reasoning effort; " "requested '%s' treated as default thinking", self.tokenizer_manager.server_args.served_model_name, request.reasoning_effort, ) # Split apply_chat_template(tokenize=True) into render + encode so we # can skip add_special_tokens=False on tokenizers that don't auto-add # specials (Kimi-like, OpenAI-chat analogue of #25265). Chat # templates already include role/special tokens, so the encode must # avoid double BOS on tokenizers that would add it. encode_kwargs = ( {"add_special_tokens": False} if self._tokenizer_auto_adds_specials else {} ) try: rendered_prompt = self.tokenizer_manager.tokenizer.apply_chat_template( openai_compatible_messages, tokenize=False, add_generation_prompt=True, tools=tools, return_dict=False, **extra_template_kwargs, ) prompt_ids = self.tokenizer_manager.tokenizer.encode( rendered_prompt, **encode_kwargs ) except Exception: # If the first attempt fails, try with flat function-only format. # Some templates (e.g. Mistral) expect tools without the OpenAI wrapper. tools = ( [t["function"] if "function" in t else t for t in tools] if tools else None ) try: rendered_prompt = ( self.tokenizer_manager.tokenizer.apply_chat_template( openai_compatible_messages, tokenize=False, add_generation_prompt=True, tools=tools, return_dict=False, **extra_template_kwargs, ) ) prompt_ids = self.tokenizer_manager.tokenizer.encode( rendered_prompt, **encode_kwargs ) except (jinja2.TemplateError, TypeError) as template_error: # Template errors (e.g., from raise_exception in Jinja templates) # and TypeError (e.g., tojson filter on Jinja2 Undefined variables) # should be treated as client errors (400 BadRequest) raise ValueError(str(template_error)) from template_error # Append assistant prefix if continue_final_message is enabled if assistant_prefix: prompt_ids = self._append_assistant_prefix_to_prompt_ids( prompt_ids, assistant_prefix ) if is_multimodal: prompt = self.tokenizer_manager.tokenizer.decode(prompt_ids) stop = request.stop image_data = image_data if image_data else None audio_data = audio_data if audio_data else None video_data = video_data if video_data else None modalities = modalities if modalities else [] return MessageProcessingResult( prompt=prompt, prompt_ids=prompt_ids, image_data=image_data, video_data=video_data, audio_data=audio_data, modalities=modalities, stop=stop, ) def _apply_conversation_template( self, request: ChatCompletionRequest, is_multimodal: bool, ) -> MessageProcessingResult: """Apply conversation template""" prompt = "" prompt_ids = [] conv = generate_chat_conv(request, self.template_manager.chat_template_name) # If we should continue the final assistant message, adjust the conversation. if ( request.continue_final_message and request.messages and request.messages[-1].role == "assistant" ): # Remove the auto-added blank assistant turn, if present. if conv.messages and conv.messages[-1][1] is None: conv.messages.pop() # Rebuild the prompt from the conversation. prompt = conv.get_prompt() # Strip trailing stop tokens or separators that indicate end-of-assistant. if isinstance(conv.stop_str, list): for stop_token in conv.stop_str: if prompt.endswith(stop_token): prompt = prompt[: -len(stop_token)] elif isinstance(conv.stop_str, str) and prompt.endswith(conv.stop_str): prompt = prompt[: -len(conv.stop_str)] if conv.sep and prompt.endswith(conv.sep): prompt = prompt[: -len(conv.sep)] if getattr(conv, "sep2", None) and prompt.endswith(conv.sep2): prompt = prompt[: -len(conv.sep2)] else: prompt = conv.get_prompt() if self._get_reasoning_from_request(request) and ( self._reasoning_detector is None or not self._reasoning_detector.thinks_internally ): # Models with thinks_internally=True think without a leading token prompt += "" # Note(Xinyuan): hard code thinking token image_data = conv.image_data if conv.image_data else None video_data = conv.video_data if conv.video_data else None audio_data = conv.audio_data if conv.audio_data else None modalities = conv.modalities if conv.modalities else [] stop = copy.copy(conv.stop_str or [] if not request.ignore_eos else []) if request.stop: if isinstance(request.stop, str): stop.append(request.stop) else: stop.extend(request.stop) if not is_multimodal: prompt_ids = self.tokenizer_manager.tokenizer.encode(prompt) return MessageProcessingResult( prompt=prompt, prompt_ids=prompt_ids, image_data=image_data, video_data=video_data, audio_data=audio_data, modalities=modalities, stop=stop, ) async def _handle_streaming_request( self, adapted_request: GenerateReqInput, request: ChatCompletionRequest, raw_request: Request, ) -> Union[StreamingResponse, ErrorResponse]: """Handle streaming chat completion request""" generator = self._generate_chat_stream(adapted_request, request, raw_request) # Kick-start the generator to trigger validation before HTTP 200 is sent. # If validation fails (e.g., context length exceeded), we can still return # a proper HTTP 400 error response instead of streaming it as SSE payload. try: first_chunk = await generator.__anext__() except ValueError as e: return self.create_error_response(str(e)) async def prepend_first_chunk(): yield first_chunk async for chunk in generator: yield chunk return StreamingResponse( prepend_first_chunk(), media_type="text/event-stream", background=self.tokenizer_manager.create_abort_task(adapted_request), ) async def _generate_chat_stream( self, adapted_request: GenerateReqInput, request: ChatCompletionRequest, raw_request: Request, ) -> AsyncGenerator[str, None]: """Generate streaming chat completion response""" # Parsers for tool calls and reasoning parser_dict = {} reasoning_parser_dict = {} # State tracking for streaming is_firsts = {} stream_offsets = {} n_prev_tokens = {} has_tool_calls = {} finish_reasons = {} # Usage tracking prompt_tokens = {} reasoning_tokens = {} completion_tokens = {} cached_tokens = {} hidden_states = {} routed_experts = {} cached_tokens_details = {} image_tokens = {} audio_tokens = {} video_tokens = {} stream_started = False try: include_usage, continuous_usage_stats = should_include_usage( request.stream_options, self.tokenizer_manager.server_args.stream_response_default_include_usage, ) async for content in self.tokenizer_manager.generate_request( adapted_request, raw_request ): index = content.get("index", 0) prompt_tokens[index] = content["meta_info"].get("prompt_tokens", 0) completion_tokens[index] = content["meta_info"].get( "completion_tokens", 0 ) reasoning_tokens[index] = content["meta_info"].get( "reasoning_tokens", 0 ) cached_tokens[index] = content["meta_info"].get("cached_tokens", 0) hidden_states[index] = content["meta_info"].get("hidden_states", None) routed_experts[index] = content["meta_info"].get("routed_experts", None) cached_tokens_details[index] = content["meta_info"].get( "cached_tokens_details", None ) image_tokens[index] = content["meta_info"].get("image_tokens", 0) audio_tokens[index] = content["meta_info"].get("audio_tokens", 0) video_tokens[index] = content["meta_info"].get("video_tokens", 0) # Handle logprobs choice_logprobs = None if request.logprobs: n_prev_token = n_prev_tokens.get(index, 0) total_output_logprobs = content["meta_info"][ "output_token_logprobs_length" ] if n_prev_token < total_output_logprobs: choice_logprobs = self._process_streaming_logprobs( content, n_prev_token, total_output_logprobs ).model_dump() n_prev_tokens[index] = total_output_logprobs finish_reason = content["meta_info"].get("finish_reason", None) finish_reason_type = finish_reason["type"] if finish_reason else None # Track finish_reason for each index if finish_reason_type: # Abort with an explicit error status_code is a system error # (timeout, OOM, validation): emit a streaming error chunk. # A graceful abort (no status_code, e.g. user-initiated via # /abort_request or session lifecycle cleanup) falls through # to the normal chunk path, matching the non-stream behavior # in tokenizer_manager._handle_abort_finish_reason. if finish_reason_type == "abort" and isinstance( finish_reason.get("status_code"), HTTPStatus ): code = finish_reason["status_code"] error = self.create_streaming_error_response( finish_reason.get("message", "Generation aborted."), code.name, code.value, ) yield f"data: {error}\n\n" break finish_reasons[index] = finish_reason # First chunk with role if is_firsts.get(index, True): is_firsts[index] = False yield build_sse_content( chunk_id=content["meta_info"]["id"], created=int(time.time()), model=request.model, index=index, role="assistant", content="", ) stream_started = True # Generate streaming content (override in subclass for custom behavior) async for chunk in self._generate_stream_content( content=content, index=index, request=request, stream_offsets=stream_offsets, reasoning_parser_dict=reasoning_parser_dict, parser_dict=parser_dict, has_tool_calls=has_tool_calls, choice_logprobs=choice_logprobs, finish_reason_type=finish_reason_type, continuous_usage_stats=continuous_usage_stats, prompt_tokens=prompt_tokens, reasoning_tokens=reasoning_tokens, completion_tokens=completion_tokens, ): yield chunk # Send finish_reason chunks for each index that completed for idx, finish_reason_data in finish_reasons.items(): finish_reason_type = finish_reason_data["type"] # Change finish_reason to "tool_calls" if we had tool calls and stopped naturally final_finish_reason = finish_reason_type if has_tool_calls.get(idx, False) and finish_reason_type == "stop": final_finish_reason = "tool_calls" matched_stop = finish_reason_data.get("matched") yield build_sse_content( chunk_id=content["meta_info"]["id"], created=int(time.time()), model=request.model, index=idx, finish_reason=final_finish_reason, matched_stop=matched_stop, ) # Send hidden states if requested if request.return_hidden_states and hidden_states: for index, choice_hidden_states in hidden_states.items(): if choice_hidden_states: last_token_hidden_states = ( choice_hidden_states[-1] if len(choice_hidden_states) > 1 else [] ) hidden_states_chunk = ChatCompletionStreamResponse( id=content["meta_info"]["id"], created=int(time.time()), choices=[ ChatCompletionResponseStreamChoice( index=index, delta=DeltaMessage( hidden_states=last_token_hidden_states ), finish_reason=None, # Hidden states don't need finish_reason ) ], model=request.model, ) yield f"data: {hidden_states_chunk.model_dump_json()}\n\n" sglext_routed = None if request.return_routed_experts and routed_experts: sglext_routed = next( (v for v in routed_experts.values() if v is not None), None ) sglext_details = None if request.return_cached_tokens_details and cached_tokens_details: first_details = next( (v for v in cached_tokens_details.values() if v is not None), None ) if first_details is not None: sglext_details = cached_tokens_details_from_dict(first_details) if sglext_routed is not None or sglext_details is not None: sglext_chunk = ChatCompletionStreamResponse( id=content["meta_info"]["id"], created=int(time.time()), choices=[], # sglext is at response level model=request.model, sglext=SglExt( routed_experts=sglext_routed, cached_tokens_details=sglext_details, ), ) yield f"data: {sglext_chunk.model_dump_json()}\n\n" # Additional usage chunk if include_usage: # Multimodal tokens are per-prompt (input side), so aggregate # once per prompt (first choice), matching prompt/cached semantics. total_image_tokens = sum( tok for idx, tok in image_tokens.items() if idx % request.n == 0 ) total_audio_tokens = sum( tok for idx, tok in audio_tokens.items() if idx % request.n == 0 ) total_video_tokens = sum( tok for idx, tok in video_tokens.items() if idx % request.n == 0 ) usage = UsageProcessor.calculate_streaming_usage( prompt_tokens, reasoning_tokens, completion_tokens, cached_tokens=cached_tokens, n_choices=request.n, enable_cache_report=self.tokenizer_manager.server_args.enable_cache_report, image_tokens=total_image_tokens, audio_tokens=total_audio_tokens, video_tokens=total_video_tokens, ) usage_chunk = ChatCompletionStreamResponse( id=content["meta_info"]["id"], created=int(time.time()), choices=[], # Empty choices array as per OpenAI spec model=request.model, usage=usage, ) yield f"data: {usage_chunk.model_dump_json()}\n\n" except ValueError as e: if not stream_started: raise error = self.create_streaming_error_response(str(e)) yield f"data: {error}\n\n" yield "data: [DONE]\n\n" async def _handle_non_streaming_request( self, adapted_request: GenerateReqInput, request: ChatCompletionRequest, raw_request: Request, ) -> Union[ChatCompletionResponse, ErrorResponse, ORJSONResponse]: """Handle non-streaming chat completion request""" try: ret = await self.tokenizer_manager.generate_request( adapted_request, raw_request ).__anext__() except ValueError as e: return self.create_error_response(str(e)) if not isinstance(ret, list): ret = [ret] response = self._build_chat_response( request, ret, int(time.time()), ) return response def _build_chat_response( self, request: ChatCompletionRequest, ret: List[Dict[str, Any]], created: int, ) -> Union[ChatCompletionResponse, ORJSONResponse]: """Build chat completion response from generation results""" choices = [] # Build sglext at response level (from first ret_item, as these are per-request) first_ret = ret[0] routed_experts = process_routed_experts_from_ret(first_ret, request) cached_tokens_details = process_cached_tokens_details_from_ret( first_ret, request ) response_sglext = None if routed_experts or cached_tokens_details: response_sglext = SglExt( routed_experts=routed_experts, cached_tokens_details=cached_tokens_details, ) for idx, ret_item in enumerate(ret): # Process logprobs choice_logprobs = None if request.logprobs: choice_logprobs = self._process_response_logprobs(ret_item) # Handle hidden states hidden_states = process_hidden_states_from_ret(ret_item, request) finish_reason = ret_item["meta_info"]["finish_reason"] text = self._decode_response(ret_item) if isinstance(text, ErrorResponse): return ORJSONResponse(content=text.model_dump(), status_code=text.code) # Handle reasoning content reasoning_text = None if self.reasoning_parser and request.separate_reasoning: force_reasoning = ( self.template_manager.force_reasoning or self._get_reasoning_from_request(request) ) try: parser = ReasoningParser( model_type=self.reasoning_parser, stream_reasoning=False, force_reasoning=force_reasoning, request=request, tokenizer=self.tokenizer_manager.tokenizer, ) reasoning_text, text = parser.parse_non_stream(text) except Exception as e: logger.error(f"Reasoning parsing error: {e}") return self.create_error_response( "Failed to parse reasoning content", err_type="InternalServerError", status_code=500, ) # Handle tool calls tool_calls = None if ( request.tool_choice != "none" and request.tools and self.tool_call_parser ): history_tool_calls_cnt = self._get_history_tool_calls_cnt(request) tool_calls, text, finish_reason = self._process_tool_calls( text, request.tools, finish_reason, request.tool_choice, history_tool_calls_cnt, ) # Extract prompt_token_ids if requested choice_prompt_token_ids = ( ret_item.get("prompt_token_ids") if request.return_prompt_token_ids else None ) choice_meta_info = ( ret_item["meta_info"] if request.return_meta_info else None ) # NOTE: content should not be None but empty string to make sure retokenize consistency. reasoning_text, tool_calls = self._get_parsed_response_fields( reasoning_text, tool_calls ) choice_data = ChatCompletionResponseChoice( index=idx, message=ChatMessage( role="assistant", content=text if text else "", tool_calls=tool_calls, reasoning_content=reasoning_text if reasoning_text else None, ), logprobs=choice_logprobs, finish_reason=finish_reason["type"] if finish_reason else None, matched_stop=( finish_reason["matched"] if finish_reason and "matched" in finish_reason else None ), hidden_states=hidden_states, prompt_token_ids=choice_prompt_token_ids, meta_info=choice_meta_info, ) choices.append(choice_data) # Calculate usage. Multimodal tokens are per-prompt (input side), so # aggregate once per prompt (stride by n), matching prompt/cached semantics. image_tokens = sum( ret[i]["meta_info"].get("image_tokens", 0) for i in range(0, len(ret), request.n) ) audio_tokens = sum( ret[i]["meta_info"].get("audio_tokens", 0) for i in range(0, len(ret), request.n) ) video_tokens = sum( ret[i]["meta_info"].get("video_tokens", 0) for i in range(0, len(ret), request.n) ) usage = UsageProcessor.calculate_response_usage( ret, n_choices=request.n, enable_cache_report=self.tokenizer_manager.server_args.enable_cache_report, image_tokens=image_tokens, audio_tokens=audio_tokens, video_tokens=video_tokens, ) return ChatCompletionResponse( id=ret[0]["meta_info"]["id"], created=created, model=request.model, choices=choices, usage=usage, metadata={"weight_version": ret[0]["meta_info"]["weight_version"]}, sglext=response_sglext, ) def _process_logprobs_tokens( self, logprobs: LogProbs, use_token_index: bool = False ) -> List[ChatCompletionTokenLogprob]: """Common helper to process logprobs tokens for both streaming and non-streaming Args: logprobs: LogProbs data from model use_token_index: True for non-streaming (use token_idx), False for streaming (use index 0) """ token_logprobs = [] for token_idx, (token, logprob) in enumerate( zip(logprobs.tokens, logprobs.token_logprobs) ): token_bytes = list(token.encode("utf-8")) top_logprobs = [] if logprobs.top_logprobs: # - Non-streaming (use_token_index=True): uses token_idx for full data # - Streaming (use_token_index=False): uses index 0 for pre-sliced data top_logprobs_idx = token_idx if use_token_index else 0 for top_token, top_logprob in logprobs.top_logprobs[ top_logprobs_idx ].items(): top_token_bytes = list(top_token.encode("utf-8")) top_logprobs.append( TopLogprob( token=top_token, bytes=top_token_bytes, logprob=top_logprob, ) ) token_logprobs.append( ChatCompletionTokenLogprob( token=token, bytes=token_bytes, logprob=logprob, top_logprobs=top_logprobs, ) ) return token_logprobs def _process_response_logprobs(self, ret_item: Dict[str, Any]) -> ChoiceLogprobs: """Process logprobs for non-streaming response""" logprobs = to_openai_style_logprobs( output_token_logprobs=ret_item["meta_info"]["output_token_logprobs"], output_top_logprobs=ret_item["meta_info"].get("output_top_logprobs", None), ) token_logprobs = self._process_logprobs_tokens(logprobs, use_token_index=True) return ChoiceLogprobs(content=token_logprobs) def _process_tool_call_id( self, call_item: ToolCallItem, history_tool_calls_cnt: int, ) -> str: """Process for generating a new and unique `tool_call_id`""" if self.tool_call_parser != "kimi_k2": # A simple uuid is sufficient for all models except for Kimi-K2. tool_call_id = f"call_{uuid.uuid4().hex[:24]}" return tool_call_id else: # Align with Kimi-K2 format: functions.{name}:{index} # Kimi-K2 allows multiple tool_calls in one message; SGLang sets call_item.tool_index to the *local* position inside that message. # Therefore, the index must be corrected by using `history_tool_calls_cnt + call_item.tool_index` to ensure globally unique and properly ordered. tool_call_id = f"functions.{call_item.name}:{history_tool_calls_cnt+call_item.tool_index}" logger.debug( f"Process tool call idx, parser: {self.tool_call_parser}, tool_call_id: {tool_call_id}, history_cnt: {history_tool_calls_cnt}" ) return tool_call_id def _process_tool_calls( self, text: str, tools: List[Any], finish_reason: Dict[str, Any], tool_choice: Optional[Union[str, ToolChoice]] = None, history_tool_calls_cnt: int = 0, ) -> ToolCallProcessingResult: """Process tool calls in the response""" is_required = tool_choice == "required" or isinstance(tool_choice, ToolChoice) # Try model-specific parser when output is in native format. # For required/named: only use parser when structural_tag was used # as constraint (mirrors the streaming path). For auto: always try. if self.tool_call_parser: parser = FunctionCallParser( tools, self.tool_call_parser, tokenizer=self.tokenizer_manager.tokenizer ) should_try_parser = ( not is_required or parser.detector.supports_structural_tag() ) if should_try_parser and parser.has_tool_call(text): original_finish_type = finish_reason["type"] if finish_reason["type"] == "stop": finish_reason["type"] = "tool_calls" finish_reason["matched"] = None try: text, call_info_list = parser.parse_non_stream(text) tool_calls = [] for call_info in call_info_list: tool_id = self._process_tool_call_id( call_info, history_tool_calls_cnt ) tool_calls.append( ToolCall( id=tool_id, index=getattr(call_info, "tool_index", None), function=FunctionResponse( name=call_info.name, arguments=call_info.parameters, ), ) ) return ToolCallProcessingResult(tool_calls, text, finish_reason) except Exception as e: logger.error(f"Tool call parsing error: {e}") finish_reason["type"] = original_finish_type return ToolCallProcessingResult(None, text, finish_reason) # json_schema constraint → JSON array output for required/named if is_required: original_finish_type = finish_reason["type"] if finish_reason["type"] == "stop": finish_reason["type"] = "tool_calls" finish_reason["matched"] = None try: tool_call_data = orjson.loads(text) tool_calls = [] for i, tool in enumerate(tool_call_data): call_info = ToolCallItem( tool_index=i, name=tool["name"], parameters=json.dumps(tool["parameters"], ensure_ascii=False), ) tool_id = self._process_tool_call_id( call_info, history_tool_calls_cnt ) tool_calls.append( ToolCall( id=tool_id, index=i, function=FunctionResponse( name=tool["name"], arguments=json.dumps( tool["parameters"], ensure_ascii=False ), ), ) ) return ToolCallProcessingResult(tool_calls, "", finish_reason) except Exception as e: logger.error(f"Tool call parsing error: {e}") finish_reason["type"] = original_finish_type return ToolCallProcessingResult(None, text, finish_reason) return ToolCallProcessingResult(None, text, finish_reason) def _process_streaming_logprobs( self, content: Dict[str, Any], n_prev_token: int, total_output_logprobs: int, ) -> ChoiceLogprobs: """Process logprobs for streaming response""" output_token_logprobs = content["meta_info"]["output_token_logprobs"] output_top_logprobs = content["meta_info"].get("output_top_logprobs", []) if not self.tokenizer_manager.server_args.incremental_streaming_output: output_token_logprobs = output_token_logprobs[ n_prev_token:total_output_logprobs ] output_top_logprobs = output_top_logprobs[ n_prev_token:total_output_logprobs ] logprobs = to_openai_style_logprobs( output_token_logprobs=output_token_logprobs, output_top_logprobs=output_top_logprobs, ) token_logprobs = self._process_logprobs_tokens(logprobs, use_token_index=False) return ChoiceLogprobs(content=token_logprobs) def _process_reasoning_stream( self, index: int, delta: str, reasoning_parser_dict: Dict[int, ReasoningParser], content: Dict[str, Any], request: ChatCompletionRequest, ) -> tuple[Optional[str], str]: """Process reasoning content in streaming response""" if index not in reasoning_parser_dict: is_force_reasoning = ( self.template_manager.force_reasoning or self._get_reasoning_from_request(request) ) reasoning_parser_dict[index] = ReasoningParser( self.reasoning_parser, request.stream_reasoning, is_force_reasoning, request, tokenizer=self.tokenizer_manager.tokenizer, ) reasoning_parser = reasoning_parser_dict[index] return reasoning_parser.parse_stream_chunk(delta) def _get_history_tool_calls_cnt(self, request: ChatCompletionRequest) -> int: """Counts the number of tool calls in the request's message history. NOTE: This method is only useful for models that include self-increasing history tool call idx in tool calls id, such as kimi-k2 Args: request: The chat completion request object. Returns: The total number of tool calls in the history, or 0 if not applicable. """ messages = getattr(request, "messages", []) idx = 0 for msg in messages: if msg.role == "assistant": tool_calls = getattr(msg, "tool_calls", None) idx += len(list(tool_calls)) if tool_calls is not None else 0 # noqa return idx def _patch_reasoning_skip_special_tokens( self, request: ChatCompletionRequest ) -> None: """Keep parser-specific reasoning markers in the decoded text. Some reasoning parsers rely on special-token delimiters that would be removed during detokenization when ``skip_special_tokens=True``. """ if self.reasoning_parser == "apertus2509": request.skip_special_tokens = False if ( self.reasoning_parser in ["mistral"] and request.reasoning_effort is not None and request.reasoning_effort != "none" ): request.skip_special_tokens = False def wrap_reasoning_history(self, reasoning_text: str) -> str: """Wrap prior-turn reasoning in the detector's own start/end tokens. Pulling the delimiters from the detector keeps adapters in lockstep with any future parser that ships non-```` markers — Mistral's ``[THINK]``, Gemma4's ``think_start_self_label = "thought\\n"``, etc. Falling back to a plain string is unsafe: it would let prior thinking text reach a non-reasoning model as ordinary assistant content, so the caller must surface this state, not paper over it. """ if self._reasoning_detector is None: raise ValueError( "Cannot rewrap thinking history: no reasoning detector is " "configured for this model" ) d = self._reasoning_detector return ( f"{d.think_start_token}{d.think_start_self_label}" f"{reasoning_text}\n{d.think_end_token}" ) def _reasoning_default_mode(self) -> Optional[str]: if self._reasoning_detector is None: return None return self._reasoning_detector.reasoning_default def _get_reasoning_toggle_param(self) -> Optional[str]: """Resolve the chat-template kwarg that toggles reasoning, if any.""" config = self.template_manager.reasoning_config if config is not None: return config.toggle_param mode = self._reasoning_default_mode() if mode in ("thinking", "enable_thinking"): return mode if mode in ("explicit_thinking", "explicit_enable_thinking"): return mode.replace("explicit_", "") return None def apply_reasoning_enabled( self, request: ChatCompletionRequest, enabled: bool ) -> None: """Force the request into the requested reasoning-on/off mode. Mirrors the read-side logic in ``_get_reasoning_from_request``; the two must stay in sync. Always-on models cannot be disabled, so explicit ``enabled=False`` raises rather than silently leaving reasoning on. """ if not self.reasoning_parser: if enabled: raise ValueError( "Anthropic thinking is not supported for models without " "a reasoning parser" ) return if self.reasoning_parser == "hunyuan": request.reasoning_effort = "medium" if enabled else "no_think" return config = self.template_manager.reasoning_config is_mistral = (config is not None and config.special_case == "mistral") or ( config is None and self._reasoning_default_mode() == "mistral" ) if is_mistral: request.reasoning_effort = "medium" if enabled else "none" return is_always_on = (config is not None and config.special_case == "always") or ( config is None and self._reasoning_default_mode() == "always" ) if is_always_on: if not enabled: raise ValueError( f"Reasoning parser '{self.reasoning_parser}' is always-on " f"and cannot be disabled via Anthropic thinking" ) return toggle_param = self._get_reasoning_toggle_param() # The read side (``_get_reasoning_from_request``) returns False # whenever ``config.toggle_param is None`` OR # ``config.default_enabled is None``. The write side must mirror # both conditions: if ``default_enabled`` is unset we cannot # actually honor an ``enabled=True`` request even when the toggle # name itself is resolvable, so writing the kwarg would set up the # template to emit reasoning tokens while the parser ignores them # (literal ```` markers leak into the assistant text). config = self.template_manager.reasoning_config read_side_supported = toggle_param is not None and ( config is None or config.default_enabled is not None ) if not read_side_supported: if not enabled: return raise ValueError( f"Anthropic thinking is not supported for reasoning parser " f"'{self.reasoning_parser}'" ) chat_template_kwargs = dict(request.chat_template_kwargs or {}) chat_template_kwargs[toggle_param] = enabled request.chat_template_kwargs = chat_template_kwargs def _get_reasoning_from_request(self, request: ChatCompletionRequest) -> bool: """Determine whether reasoning mode should be enabled for this request. NOTE: This is predefined based on model's chat template """ if not self.reasoning_parser: return False if self.reasoning_parser == "minimax-m3": # M3 template prefills for thinking_mode=enabled, so it never # appears in output and reasoning must be forced. Mirrors reasoning_parser.py. return (request.chat_template_kwargs or {}).get( "thinking_mode" ) == "enabled" if self.reasoning_parser == "hunyuan": # Hy3-preview template emits no when reasoning_effort is # "no_think" / "none" / unset; forcing reasoning would route all # output into reasoning_content. return request.reasoning_effort not in (None, "none", "no_think") config = self.template_manager.reasoning_config if config is None: # Fallback to parser-level defaults when template toggle config # cannot be inferred (e.g., parser-only templates). mode = ( self._reasoning_detector.reasoning_default if self._reasoning_detector is not None else None ) if mode is None: return False if mode == "always": return True if mode == "mistral": return ( request.reasoning_effort is not None and request.reasoning_effort != "none" ) if mode in ("thinking", "enable_thinking"): return ( not request.chat_template_kwargs or request.chat_template_kwargs.get(mode) is not False ) if mode in ("explicit_thinking", "explicit_enable_thinking"): toggle = mode.replace("explicit_", "") return ( request.chat_template_kwargs is not None and request.chat_template_kwargs.get(toggle) is True ) logger.warning( "Unknown reasoning_default mode '%s', defaulting to reasoning disabled", mode, ) return False if config.special_case == "always": return True if config.special_case == "mistral": return ( request.reasoning_effort is not None and request.reasoning_effort != "none" ) if config.toggle_param is None or config.default_enabled is None: return False if config.default_enabled: return ( not request.chat_template_kwargs or request.chat_template_kwargs.get(config.toggle_param) is not False ) return ( request.chat_template_kwargs is not None and request.chat_template_kwargs.get(config.toggle_param) is True ) async def _process_tool_call_stream( self, index: int, delta: str, parser_dict: Dict[int, FunctionCallParser], content: Dict[str, Any], request: ChatCompletionRequest, has_tool_calls: Dict[int, bool], continuous_usage_stats: bool = False, ): """Process tool calls in streaming response""" if index not in parser_dict: is_required = request.tool_choice == "required" or isinstance( request.tool_choice, ToolChoice ) # For required/named tool choice: use JsonArrayParser when the # constrained output is plain JSON (detector doesn't support # structural_tag or no parser configured). Use FunctionCallParser # only when the detector supports structural_tag and will produce # native format output. if is_required: use_native_parser = False if self.tool_call_parser: probe = FunctionCallParser( tools=request.tools, tool_call_parser=self.tool_call_parser, tokenizer=self.tokenizer_manager.tokenizer, ) use_native_parser = probe.detector.supports_structural_tag() if use_native_parser: parser_dict[index] = probe else: parser_dict[index] = JsonArrayParser() else: parser_dict[index] = FunctionCallParser( tools=request.tools, tool_call_parser=self.tool_call_parser, tokenizer=self.tokenizer_manager.tokenizer, ) parser = parser_dict[index] # Handle both FunctionCallParser and JsonArrayParser if isinstance(parser, JsonArrayParser): result = parser.parse_streaming_increment(delta, request.tools) normal_text, calls = result.normal_text, result.calls else: normal_text, calls = parser.parse_stream_chunk(delta) # Yield normal text if normal_text: choice_data = ChatCompletionResponseStreamChoice( index=index, delta=DeltaMessage(content=normal_text), finish_reason=None, ) chunk = ChatCompletionStreamResponse( id=content["meta_info"]["id"], created=int(time.time()), choices=[choice_data], model=request.model, ) # Add usage stats if continuous_usage_stats is enabled if continuous_usage_stats: prompt_tokens = content["meta_info"].get("prompt_tokens", 0) completion_tokens = content["meta_info"].get("completion_tokens", 0) reasoning_tokens = content["meta_info"].get("reasoning_tokens", 0) chunk.usage = UsageProcessor.calculate_token_usage( prompt_tokens=prompt_tokens, completion_tokens=completion_tokens, reasoning_tokens=reasoning_tokens, cached_tokens=self._continuous_usage_cached_details(content), ) yield f"data: {chunk.model_dump_json()}\n\n" # Yield tool calls history_tool_calls_cnt = self._get_history_tool_calls_cnt(request) for call_item in calls: # Mark that this choice has tool calls has_tool_calls[index] = True # Tool call ID should be generated only once per tool call if call_item.name: # First chunk: include ID and function name tool_call_id = self._process_tool_call_id( call_item, history_tool_calls_cnt ) function_name = call_item.name else: # Subsequent chunks: null ID and name for argument deltas tool_call_id = None function_name = None tool_call = ToolCall( id=tool_call_id, index=call_item.tool_index, function=FunctionResponse( name=function_name, arguments=call_item.parameters, ), ) choice_data = ChatCompletionResponseStreamChoice( index=index, delta=DeltaMessage(tool_calls=[tool_call]), finish_reason=None, ) chunk = ChatCompletionStreamResponse( id=content["meta_info"]["id"], created=int(time.time()), choices=[choice_data], model=request.model, ) # Add usage stats if continuous_usage_stats is enabled if continuous_usage_stats: prompt_tokens = content["meta_info"].get("prompt_tokens", 0) completion_tokens = content["meta_info"].get("completion_tokens", 0) reasoning_tokens = content["meta_info"].get("reasoning_tokens", 0) chunk.usage = UsageProcessor.calculate_token_usage( prompt_tokens=prompt_tokens, completion_tokens=completion_tokens, reasoning_tokens=reasoning_tokens, cached_tokens=self._continuous_usage_cached_details(content), ) yield f"data: {chunk.model_dump_json()}\n\n" def _check_for_unstreamed_tool_args( self, parser: Union[FunctionCallParser, JsonArrayParser], content: Dict[str, Any], request: ChatCompletionRequest, index: int, ) -> Optional[str]: """ Check for any remaining tool call arguments that need to be streamed when generation finishes. This ensures tool calls are properly completed even if the model generates the final arguments in the last chunk. """ # Get the detector - either from FunctionCallParser or directly if json detector detector = parser.detector if hasattr(parser, "detector") else parser # Only check if we have tool calls and the detector has tracked data if ( not hasattr(detector, "prev_tool_call_arr") or not detector.prev_tool_call_arr ): return None if ( not hasattr(detector, "streamed_args_for_tool") or not detector.streamed_args_for_tool ): return None # Get the last tool call that was being processed tool_index = len(detector.prev_tool_call_arr) - 1 if tool_index < 0 or tool_index >= len(detector.streamed_args_for_tool): return None # Get expected vs actual arguments expected_args = detector.prev_tool_call_arr[tool_index].get("arguments", {}) if isinstance(expected_args, str): expected_call = expected_args else: expected_call = json.dumps(expected_args, ensure_ascii=False) actual_call = detector.streamed_args_for_tool[tool_index] # Check if there are remaining arguments to send remaining_call = ( expected_call[len(actual_call) :] if expected_call.startswith(actual_call) else "" ) if remaining_call: # Create tool call chunk with remaining arguments tool_call = ToolCall( id=None, # No ID for argument deltas index=tool_index, function=FunctionResponse( name=None, # No name for argument deltas arguments=remaining_call, ), ) choice_data = ChatCompletionResponseStreamChoice( index=index, delta=DeltaMessage(tool_calls=[tool_call]), finish_reason=None, # Don't send finish_reason with this chunk ) chunk = ChatCompletionStreamResponse( id=content["meta_info"]["id"], created=int(time.time()), choices=[choice_data], model=request.model, ) return f"data: {chunk.model_dump_json()}\n\n" return None