From 36c4745bb6954b802e727be1bbd3f2331c19d0f0 Mon Sep 17 00:00:00 2001 From: Tai An Date: Tue, 14 Apr 2026 23:59:12 -0700 Subject: [PATCH] fix(openai): pass param for all OpenAI-compatible endpoints Previously, the field was only added to supported params when the model name appeared in the official litellm model lists. This silently dropped for any custom OpenAI-compatible endpoint (e.g. local proxies, self-hosted models), with no warning or error. is a standard field in the OpenAI Chat Completions API spec and should be forwarded for all endpoints using the OpenAI provider. Azure AI already has its own transformation class that handles independently. Fixes #25753 --- .../llms/openai/chat/gpt_transformation.py | 1636 ++++++++--------- 1 file changed, 816 insertions(+), 820 deletions(-) diff --git a/litellm/llms/openai/chat/gpt_transformation.py b/litellm/llms/openai/chat/gpt_transformation.py index c12c6e6ba09..6d227823984 100644 --- a/litellm/llms/openai/chat/gpt_transformation.py +++ b/litellm/llms/openai/chat/gpt_transformation.py @@ -1,820 +1,816 @@ -""" -Support for gpt model family -""" - -from typing import ( - TYPE_CHECKING, - Any, - AsyncIterator, - Coroutine, - Dict, - Iterator, - List, - Literal, - Optional, - Tuple, - Union, - cast, - overload, -) - -import httpx - -import litellm -from litellm.litellm_core_utils.core_helpers import map_finish_reason -from litellm.litellm_core_utils.llm_response_utils.convert_dict_to_response import ( - _extract_reasoning_content, - _handle_invalid_parallel_tool_calls, - _should_convert_tool_call_to_json_mode, -) -from litellm.litellm_core_utils.prompt_templates.common_utils import get_tool_call_names -from litellm.litellm_core_utils.prompt_templates.image_handling import ( - async_convert_url_to_base64, - convert_url_to_base64, -) -from litellm.llms.base_llm.base_model_iterator import BaseModelResponseIterator -from litellm.llms.base_llm.base_utils import BaseLLMModelInfo -from litellm.llms.base_llm.chat.transformation import BaseConfig, BaseLLMException -from litellm.secret_managers.main import get_secret_str -from litellm.types.llms.openai import ( - AllMessageValues, - ChatCompletionFileObject, - ChatCompletionFileObjectFile, - ChatCompletionImageObject, - ChatCompletionImageUrlObject, - OpenAIChatCompletionChoices, - OpenAIMessageContentListBlock, -) -from litellm.types.utils import ( - ChatCompletionMessageToolCall, - Choices, - Function, - Message, - ModelResponse, - ModelResponseStream, -) -from litellm.utils import convert_to_model_response_object - -from ..common_utils import OpenAIError - -if TYPE_CHECKING: - from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj - from litellm.llms.base_llm.base_utils import BaseTokenCounter - from litellm.types.llms.openai import ChatCompletionToolParam - - LiteLLMLoggingObj = _LiteLLMLoggingObj -else: - LiteLLMLoggingObj = Any - - -class OpenAIGPTConfig(BaseLLMModelInfo, BaseConfig): - """ - Reference: https://platform.openai.com/docs/api-reference/chat/create - - The class `OpenAIConfig` provides configuration for the OpenAI's Chat API interface. Below are the parameters: - - - `frequency_penalty` (number or null): Defaults to 0. Allows a value between -2.0 and 2.0. Positive values penalize new tokens based on their existing frequency in the text so far, thereby minimizing repetition. - - - `function_call` (string or object): This optional parameter controls how the model calls functions. - - - `functions` (array): An optional parameter. It is a list of functions for which the model may generate JSON inputs. - - - `logit_bias` (map): This optional parameter modifies the likelihood of specified tokens appearing in the completion. - - - `max_tokens` (integer or null): This optional parameter helps to set the maximum number of tokens to generate in the chat completion. - - - `n` (integer or null): This optional parameter helps to set how many chat completion choices to generate for each input message. - - - `presence_penalty` (number or null): Defaults to 0. It penalizes new tokens based on if they appear in the text so far, hence increasing the model's likelihood to talk about new topics. - - - `stop` (string / array / null): Specifies up to 4 sequences where the API will stop generating further tokens. - - - `temperature` (number or null): Defines the sampling temperature to use, varying between 0 and 2. - - - `top_p` (number or null): An alternative to sampling with temperature, used for nucleus sampling. - """ - - # Add a class variable to track if this is the base class - _is_base_class = True - - frequency_penalty: Optional[int] = None - function_call: Optional[Union[str, dict]] = None - functions: Optional[list] = None - logit_bias: Optional[dict] = None - max_tokens: Optional[int] = None - n: Optional[int] = None - presence_penalty: Optional[int] = None - stop: Optional[Union[str, list]] = None - temperature: Optional[int] = None - top_p: Optional[int] = None - response_format: Optional[dict] = None - - def __init__( - self, - frequency_penalty: Optional[int] = None, - function_call: Optional[Union[str, dict]] = None, - functions: Optional[list] = None, - logit_bias: Optional[dict] = None, - max_tokens: Optional[int] = None, - n: Optional[int] = None, - presence_penalty: Optional[int] = None, - stop: Optional[Union[str, list]] = None, - temperature: Optional[int] = None, - top_p: Optional[int] = None, - response_format: Optional[dict] = None, - ) -> None: - locals_ = locals().copy() - for key, value in locals_.items(): - if key != "self" and value is not None: - setattr(self.__class__, key, value) - - self.__class__._is_base_class = False - - @classmethod - def get_config(cls): - return super().get_config() - - def get_supported_openai_params(self, model: str) -> list: - base_params = [ - "frequency_penalty", - "logit_bias", - "logprobs", - "top_logprobs", - "max_tokens", - "max_completion_tokens", - "modalities", - "prediction", - "n", - "presence_penalty", - "seed", - "stop", - "stream", - "stream_options", - "temperature", - "top_p", - "tools", - "tool_choice", - "function_call", - "functions", - "max_retries", - "extra_headers", - "parallel_tool_calls", - "audio", - "web_search_options", - "service_tier", - "safety_identifier", - "prompt_cache_key", - "prompt_cache_retention", - "store", - ] # works across all models - - model_specific_params = [] - if ( - model != "gpt-3.5-turbo-16k" and model != "gpt-4" - ): # gpt-4 does not support 'response_format' - model_specific_params.append("response_format") - - # Normalize model name for responses API (e.g., "responses/gpt-4.1" -> "gpt-4.1") - model_for_check = ( - model.split("responses/", 1)[1] if "responses/" in model else model - ) - if ( - model_for_check in litellm.open_ai_chat_completion_models - ) or model_for_check in litellm.open_ai_text_completion_models: - model_specific_params.append( - "user" - ) # user is not a param supported by all openai-compatible endpoints - e.g. azure ai - return base_params + model_specific_params - - def _map_openai_params( - self, - non_default_params: dict, - optional_params: dict, - model: str, - drop_params: bool, - ) -> dict: - """ - If any supported_openai_params are in non_default_params, add them to optional_params, so they are use in API call - - Args: - non_default_params (dict): Non-default parameters to filter. - optional_params (dict): Optional parameters to update. - model (str): Model name for parameter support check. - - Returns: - dict: Updated optional_params with supported non-default parameters. - """ - supported_openai_params = self.get_supported_openai_params(model) - for param, value in non_default_params.items(): - if param in supported_openai_params: - optional_params[param] = value - return optional_params - - def map_openai_params( - self, - non_default_params: dict, - optional_params: dict, - model: str, - drop_params: bool, - ) -> dict: - return self._map_openai_params( - non_default_params=non_default_params, - optional_params=optional_params, - model=model, - drop_params=drop_params, - ) - - def contains_pdf_url(self, content_item: ChatCompletionFileObjectFile) -> bool: - potential_pdf_url_starts = ["https://", "http://", "www."] - file_id = content_item.get("file_id") - if file_id and any( - file_id.startswith(start) for start in potential_pdf_url_starts - ): - return True - return False - - def _handle_pdf_url( - self, content_item: ChatCompletionFileObjectFile - ) -> ChatCompletionFileObjectFile: - content_copy = content_item.copy() - file_id = content_copy.get("file_id") - if file_id is not None: - base64_data = convert_url_to_base64(file_id) - content_copy["file_data"] = base64_data - content_copy["filename"] = "my_file.pdf" - content_copy.pop("file_id") - return content_copy - - async def _async_handle_pdf_url( - self, content_item: ChatCompletionFileObjectFile - ) -> ChatCompletionFileObjectFile: - file_id = content_item.get("file_id") - if file_id is not None: # check for file id being url done in _handle_pdf_url - base64_data = await async_convert_url_to_base64(file_id) - content_item["file_data"] = base64_data - content_item["filename"] = "my_file.pdf" - content_item.pop("file_id") - return content_item - - def _common_file_data_check( - self, content_item: ChatCompletionFileObjectFile - ) -> ChatCompletionFileObjectFile: - file_data = content_item.get("file_data") - filename = content_item.get("filename") - if file_data is not None and filename is None: - content_item["filename"] = "my_file.pdf" - return content_item - - def _apply_common_transform_content_item( - self, - content_item: OpenAIMessageContentListBlock, - ) -> OpenAIMessageContentListBlock: - litellm_specific_params = {"format"} - if content_item.get("type") == "image_url": - content_item = cast(ChatCompletionImageObject, content_item) - if isinstance(content_item["image_url"], str): - content_item["image_url"] = { - "url": content_item["image_url"], - } - elif isinstance(content_item["image_url"], dict): - new_image_url_obj = ChatCompletionImageUrlObject( - **{ # type: ignore - k: v - for k, v in content_item["image_url"].items() - if k not in litellm_specific_params - } - ) - content_item["image_url"] = new_image_url_obj - elif content_item.get("type") == "file": - content_item = cast(ChatCompletionFileObject, content_item) - file_obj = content_item["file"] - new_file_obj = ChatCompletionFileObjectFile( - **{ # type: ignore - k: v - for k, v in file_obj.items() - if k not in litellm_specific_params - } - ) - content_item["file"] = new_file_obj - - return content_item - - def _transform_content_item( - self, - content_item: OpenAIMessageContentListBlock, - ) -> OpenAIMessageContentListBlock: - content_item = self._apply_common_transform_content_item(content_item) - content_item_type = content_item.get("type") - potential_file_obj = content_item.get("file") - if content_item_type == "file" and potential_file_obj: - file_obj = cast(ChatCompletionFileObjectFile, potential_file_obj) - content_item_typed = cast(ChatCompletionFileObject, content_item) - if self.contains_pdf_url(file_obj): - file_obj = self._handle_pdf_url(file_obj) - file_obj = self._common_file_data_check(file_obj) - content_item_typed["file"] = file_obj - content_item = content_item_typed - return content_item - - async def _async_transform_content_item( - self, content_item: OpenAIMessageContentListBlock, is_async: bool = False - ) -> OpenAIMessageContentListBlock: - content_item = self._apply_common_transform_content_item(content_item) - content_item_type = content_item.get("type") - potential_file_obj = content_item.get("file") - if content_item_type == "file" and potential_file_obj: - file_obj = cast(ChatCompletionFileObjectFile, potential_file_obj) - content_item_typed = cast(ChatCompletionFileObject, content_item) - if self.contains_pdf_url(file_obj): - file_obj = await self._async_handle_pdf_url(file_obj) - file_obj = self._common_file_data_check(file_obj) - content_item_typed["file"] = file_obj - content_item = content_item_typed - return content_item - - # fmt: off - - @overload - def _transform_messages( - self, messages: List[AllMessageValues], model: str, is_async: Literal[True] - ) -> Coroutine[Any, Any, List[AllMessageValues]]: - ... - - @overload - def _transform_messages( - self, - messages: List[AllMessageValues], - model: str, - is_async: Literal[False] = False, - ) -> List[AllMessageValues]: - ... - - # fmt: on - - def _transform_messages( - self, messages: List[AllMessageValues], model: str, is_async: bool = False - ) -> Union[List[AllMessageValues], Coroutine[Any, Any, List[AllMessageValues]]]: - """OpenAI no longer supports image_url as a string, so we need to convert it to a dict""" - - async def _async_transform(): - for message in messages: - message_content = message.get("content") - message_role = message.get("role") - - if ( - message_role == "user" - and message_content - and isinstance(message_content, list) - ): - message_content_types = cast( - List[OpenAIMessageContentListBlock], message_content - ) - for i, content_item in enumerate(message_content_types): - message_content_types[ - i - ] = await self._async_transform_content_item( - cast(OpenAIMessageContentListBlock, content_item), - ) - return messages - - if is_async: - return _async_transform() - else: - for message in messages: - message_content = message.get("content") - message_role = message.get("role") - if ( - message_role == "user" - and message_content - and isinstance(message_content, list) - ): - message_content_types = cast( - List[OpenAIMessageContentListBlock], message_content - ) - for i, content_item in enumerate(message_content): - message_content_types[i] = self._transform_content_item( - cast(OpenAIMessageContentListBlock, content_item) - ) - return messages - - def remove_cache_control_flag_from_messages_and_tools( - self, - model: str, # allows overrides to selectively run this - messages: List[AllMessageValues], - tools: Optional[List["ChatCompletionToolParam"]] = None, - ) -> Tuple[List[AllMessageValues], Optional[List["ChatCompletionToolParam"]]]: - from litellm.litellm_core_utils.prompt_templates.common_utils import ( - filter_value_from_dict, - ) - from litellm.types.llms.openai import ChatCompletionToolParam - - for i, message in enumerate(messages): - messages[i] = cast( - AllMessageValues, filter_value_from_dict(message, "cache_control") # type: ignore - ) - if tools is not None: - for i, tool in enumerate(tools): - tools[i] = cast( - ChatCompletionToolParam, - filter_value_from_dict(tool, "cache_control"), # type: ignore - ) - return messages, tools - - def transform_request( - self, - model: str, - messages: List[AllMessageValues], - optional_params: dict, - litellm_params: dict, - headers: dict, - ) -> dict: - """ - Transform the overall request to be sent to the API. - - Returns: - dict: The transformed request. Sent as the body of the API call. - """ - messages = self._transform_messages(messages=messages, model=model) - messages, tools = self.remove_cache_control_flag_from_messages_and_tools( - model=model, messages=messages, tools=optional_params.get("tools", []) - ) - if tools is not None and len(tools) > 0: - optional_params["tools"] = tools - - optional_params.pop("max_retries", None) - - return { - "model": model, - "messages": messages, - **optional_params, - } - - async def async_transform_request( - self, - model: str, - messages: List[AllMessageValues], - optional_params: dict, - litellm_params: dict, - headers: dict, - ) -> dict: - transformed_messages = await self._transform_messages( - messages=messages, model=model, is_async=True - ) - ( - transformed_messages, - tools, - ) = self.remove_cache_control_flag_from_messages_and_tools( - model=model, - messages=transformed_messages, - tools=optional_params.get("tools", []), - ) - if tools is not None and len(tools) > 0: - optional_params["tools"] = tools - if self.__class__._is_base_class: - return { - "model": model, - "messages": transformed_messages, - **optional_params, - } - else: - ## allow for any object specific behaviour to be handled - return self.transform_request( - model, messages, optional_params, litellm_params, headers - ) - - def _passed_in_tools(self, optional_params: dict) -> bool: - return optional_params.get("tools", None) is not None - - def _check_and_fix_if_content_is_tool_call( - self, content: str, optional_params: dict - ) -> Optional[ChatCompletionMessageToolCall]: - """ - Check if the content is a tool call - """ - import json - - if not self._passed_in_tools(optional_params): - return None - tool_call_names = get_tool_call_names(optional_params.get("tools", [])) - try: - json_content = json.loads(content) - if ( - json_content.get("type") == "function" - and json_content.get("name") in tool_call_names - ): - return ChatCompletionMessageToolCall( - function=Function( - name=json_content.get("name"), - arguments=json_content.get("arguments"), - ) - ) - except Exception: - return None - - return None - - def _get_finish_reason(self, message: Message, received_finish_reason: str) -> str: - if message.tool_calls is not None: - return "tool_calls" - else: - return received_finish_reason - - def _transform_choices( - self, - choices: List[OpenAIChatCompletionChoices], - json_mode: Optional[bool] = None, - optional_params: Optional[dict] = None, - ) -> List[Choices]: - transformed_choices = [] - - for choice in choices: - ## HANDLE JSON MODE - anthropic returns single function call] - tool_calls = choice["message"].get("tool_calls", None) - new_tool_calls: Optional[List[ChatCompletionMessageToolCall]] = None - message_content = choice["message"].get("content", None) - if tool_calls is not None: - _openai_tool_calls = [] - for _tc in tool_calls: - _openai_tc = ChatCompletionMessageToolCall(**_tc) # type: ignore - _openai_tool_calls.append(_openai_tc) - fixed_tool_calls = _handle_invalid_parallel_tool_calls( - _openai_tool_calls - ) - - if fixed_tool_calls is not None: - new_tool_calls = fixed_tool_calls - elif ( - optional_params is not None - and message_content - and isinstance(message_content, str) - ): - new_tool_call = self._check_and_fix_if_content_is_tool_call( - message_content, optional_params - ) - if new_tool_call is not None: - choice["message"]["content"] = None # remove the content - new_tool_calls = [new_tool_call] - - translated_message: Optional[Message] = None - finish_reason: Optional[str] = None - if new_tool_calls and _should_convert_tool_call_to_json_mode( - tool_calls=new_tool_calls, - convert_tool_call_to_json_mode=json_mode, - ): - # to support response_format on claude models - json_mode_content_str: Optional[str] = ( - str(new_tool_calls[0]["function"].get("arguments", "")) or None - ) - if json_mode_content_str is not None: - translated_message = Message(content=json_mode_content_str) - finish_reason = "stop" - - if translated_message is None: - ## get the reasoning content - ( - reasoning_content, - content_str, - ) = _extract_reasoning_content(cast(dict, choice["message"])) - - translated_message = Message( - role="assistant", - content=content_str, - reasoning_content=reasoning_content, - thinking_blocks=None, - tool_calls=new_tool_calls, - ) - - if finish_reason is None: - finish_reason = choice["finish_reason"] - - translated_choice = Choices( - finish_reason=finish_reason, - index=choice["index"], - message=translated_message, - logprobs=None, - enhancements=None, - ) - - translated_choice.finish_reason = map_finish_reason( - self._get_finish_reason(translated_message, choice["finish_reason"]) - ) - transformed_choices.append(translated_choice) - - return transformed_choices - - def transform_response( - self, - model: str, - raw_response: httpx.Response, - model_response: ModelResponse, - logging_obj: LiteLLMLoggingObj, - request_data: dict, - messages: List[AllMessageValues], - optional_params: dict, - litellm_params: dict, - encoding: Any, - api_key: Optional[str] = None, - json_mode: Optional[bool] = None, - ) -> ModelResponse: - """ - Transform the response from the API. - - Returns: - dict: The transformed response. - """ - - ## LOGGING - logging_obj.post_call( - input=messages, - api_key=api_key, - original_response=raw_response.text, - additional_args={"complete_input_dict": request_data}, - ) - - ## RESPONSE OBJECT - try: - completion_response = raw_response.json() - except Exception as e: - response_headers = getattr(raw_response, "headers", None) - raise OpenAIError( - message="Unable to get json response - {}, Original Response: {}".format( - str(e), raw_response.text - ), - status_code=raw_response.status_code, - headers=response_headers, - ) - raw_response_headers = dict(raw_response.headers) - final_response_obj = convert_to_model_response_object( - response_object=completion_response, - model_response_object=model_response, - hidden_params={"headers": raw_response_headers}, - _response_headers=raw_response_headers, - ) - - return cast(ModelResponse, final_response_obj) - - def get_error_class( - self, error_message: str, status_code: int, headers: Union[dict, httpx.Headers] - ) -> BaseLLMException: - return OpenAIError( - status_code=status_code, - message=error_message, - headers=cast(httpx.Headers, headers), - ) - - def get_complete_url( - self, - api_base: Optional[str], - api_key: Optional[str], - model: str, - optional_params: dict, - litellm_params: dict, - stream: Optional[bool] = None, - ) -> str: - """ - Get the complete URL for the API call. - - Returns: - str: The complete URL for the API call. - """ - if api_base is None: - api_base = "https://api.openai.com" - endpoint = "chat/completions" - - # Remove trailing slash from api_base if present - api_base = api_base.rstrip("/") - - # Check if endpoint is already in the api_base - if endpoint in api_base: - return api_base - - return f"{api_base}/{endpoint}" - - def validate_environment( - self, - headers: dict, - model: str, - messages: List[AllMessageValues], - optional_params: dict, - litellm_params: dict, - api_key: Optional[str] = None, - api_base: Optional[str] = None, - ) -> dict: - if api_key is not None: - headers["Authorization"] = f"Bearer {api_key}" - - # Ensure Content-Type is set to application/json - if "content-type" not in headers and "Content-Type" not in headers: - headers["Content-Type"] = "application/json" - - return headers - - def get_models( - self, api_key: Optional[str] = None, api_base: Optional[str] = None - ) -> List[str]: - """ - Calls OpenAI's `/v1/models` endpoint and returns the list of models. - """ - - if api_base is None: - api_base = "https://api.openai.com" - if api_key is None: - api_key = get_secret_str("OPENAI_API_KEY") - - # Strip api_base to just the base URL (scheme + host + port) - parsed_url = httpx.URL(api_base) - base_url = f"{parsed_url.scheme}://{parsed_url.host}" - if parsed_url.port: - base_url += f":{parsed_url.port}" - - response = litellm.module_level_client.get( - url=f"{base_url}/v1/models", - headers={"Authorization": f"Bearer {api_key}"}, - ) - - if response.status_code != 200: - raise Exception(f"Failed to get models: {response.text}") - - models = response.json()["data"] - return [model["id"] for model in models] - - @staticmethod - def get_api_key(api_key: Optional[str] = None) -> Optional[str]: - return ( - api_key - or litellm.api_key - or litellm.openai_key - or get_secret_str("OPENAI_API_KEY") - ) - - @staticmethod - def get_api_base(api_base: Optional[str] = None) -> Optional[str]: - return ( - api_base - or litellm.api_base - or get_secret_str("OPENAI_BASE_URL") - or get_secret_str("OPENAI_API_BASE") - or "https://api.openai.com/v1" - ) - - @staticmethod - def get_base_model(model: Optional[str] = None) -> Optional[str]: - return model - - def get_token_counter(self) -> Optional["BaseTokenCounter"]: - from litellm.llms.openai.responses.count_tokens.token_counter import ( - OpenAITokenCounter, - ) - - return OpenAITokenCounter() - - def get_model_response_iterator( - self, - streaming_response: Union[Iterator[str], AsyncIterator[str], ModelResponse], - sync_stream: bool, - json_mode: Optional[bool] = False, - ) -> Any: - return OpenAIChatCompletionStreamingHandler( - streaming_response=streaming_response, - sync_stream=sync_stream, - json_mode=json_mode, - ) - - -class OpenAIChatCompletionStreamingHandler(BaseModelResponseIterator): - def _map_reasoning_to_reasoning_content(self, choices: list) -> list: - """ - Map 'reasoning' field to 'reasoning_content' field in delta. - - Some OpenAI-compatible providers (e.g., GLM-5, hosted_vllm) return - delta.reasoning, but LiteLLM expects delta.reasoning_content. - - Args: - choices: List of choice objects from the streaming chunk - - Returns: - List of choices with reasoning field mapped to reasoning_content - """ - for choice in choices: - delta = choice.get("delta", {}) - if "reasoning" in delta: - delta["reasoning_content"] = delta.pop("reasoning") - return choices - - def chunk_parser(self, chunk: dict) -> ModelResponseStream: - try: - choices = chunk.get("choices", []) - choices = self._map_reasoning_to_reasoning_content(choices) - - kwargs: Dict[str, Any] = { - "id": chunk.get("id"), - "object": "chat.completion.chunk", - "created": chunk.get("created"), - "model": chunk.get("model"), - "choices": choices, - } - if "usage" in chunk and chunk["usage"] is not None: - kwargs["usage"] = chunk["usage"] - return ModelResponseStream(**kwargs) - except Exception as e: - raise e +""" +Support for gpt model family +""" + +from typing import ( + TYPE_CHECKING, + Any, + AsyncIterator, + Coroutine, + Dict, + Iterator, + List, + Literal, + Optional, + Tuple, + Union, + cast, + overload, +) + +import httpx + +import litellm +from litellm.litellm_core_utils.core_helpers import map_finish_reason +from litellm.litellm_core_utils.llm_response_utils.convert_dict_to_response import ( + _extract_reasoning_content, + _handle_invalid_parallel_tool_calls, + _should_convert_tool_call_to_json_mode, +) +from litellm.litellm_core_utils.prompt_templates.common_utils import get_tool_call_names +from litellm.litellm_core_utils.prompt_templates.image_handling import ( + async_convert_url_to_base64, + convert_url_to_base64, +) +from litellm.llms.base_llm.base_model_iterator import BaseModelResponseIterator +from litellm.llms.base_llm.base_utils import BaseLLMModelInfo +from litellm.llms.base_llm.chat.transformation import BaseConfig, BaseLLMException +from litellm.secret_managers.main import get_secret_str +from litellm.types.llms.openai import ( + AllMessageValues, + ChatCompletionFileObject, + ChatCompletionFileObjectFile, + ChatCompletionImageObject, + ChatCompletionImageUrlObject, + OpenAIChatCompletionChoices, + OpenAIMessageContentListBlock, +) +from litellm.types.utils import ( + ChatCompletionMessageToolCall, + Choices, + Function, + Message, + ModelResponse, + ModelResponseStream, +) +from litellm.utils import convert_to_model_response_object + +from ..common_utils import OpenAIError + +if TYPE_CHECKING: + from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj + from litellm.llms.base_llm.base_utils import BaseTokenCounter + from litellm.types.llms.openai import ChatCompletionToolParam + + LiteLLMLoggingObj = _LiteLLMLoggingObj +else: + LiteLLMLoggingObj = Any + + +class OpenAIGPTConfig(BaseLLMModelInfo, BaseConfig): + """ + Reference: https://platform.openai.com/docs/api-reference/chat/create + + The class `OpenAIConfig` provides configuration for the OpenAI's Chat API interface. Below are the parameters: + + - `frequency_penalty` (number or null): Defaults to 0. Allows a value between -2.0 and 2.0. Positive values penalize new tokens based on their existing frequency in the text so far, thereby minimizing repetition. + + - `function_call` (string or object): This optional parameter controls how the model calls functions. + + - `functions` (array): An optional parameter. It is a list of functions for which the model may generate JSON inputs. + + - `logit_bias` (map): This optional parameter modifies the likelihood of specified tokens appearing in the completion. + + - `max_tokens` (integer or null): This optional parameter helps to set the maximum number of tokens to generate in the chat completion. + + - `n` (integer or null): This optional parameter helps to set how many chat completion choices to generate for each input message. + + - `presence_penalty` (number or null): Defaults to 0. It penalizes new tokens based on if they appear in the text so far, hence increasing the model's likelihood to talk about new topics. + + - `stop` (string / array / null): Specifies up to 4 sequences where the API will stop generating further tokens. + + - `temperature` (number or null): Defines the sampling temperature to use, varying between 0 and 2. + + - `top_p` (number or null): An alternative to sampling with temperature, used for nucleus sampling. + """ + + # Add a class variable to track if this is the base class + _is_base_class = True + + frequency_penalty: Optional[int] = None + function_call: Optional[Union[str, dict]] = None + functions: Optional[list] = None + logit_bias: Optional[dict] = None + max_tokens: Optional[int] = None + n: Optional[int] = None + presence_penalty: Optional[int] = None + stop: Optional[Union[str, list]] = None + temperature: Optional[int] = None + top_p: Optional[int] = None + response_format: Optional[dict] = None + + def __init__( + self, + frequency_penalty: Optional[int] = None, + function_call: Optional[Union[str, dict]] = None, + functions: Optional[list] = None, + logit_bias: Optional[dict] = None, + max_tokens: Optional[int] = None, + n: Optional[int] = None, + presence_penalty: Optional[int] = None, + stop: Optional[Union[str, list]] = None, + temperature: Optional[int] = None, + top_p: Optional[int] = None, + response_format: Optional[dict] = None, + ) -> None: + locals_ = locals().copy() + for key, value in locals_.items(): + if key != "self" and value is not None: + setattr(self.__class__, key, value) + + self.__class__._is_base_class = False + + @classmethod + def get_config(cls): + return super().get_config() + + def get_supported_openai_params(self, model: str) -> list: + base_params = [ + "frequency_penalty", + "logit_bias", + "logprobs", + "top_logprobs", + "max_tokens", + "max_completion_tokens", + "modalities", + "prediction", + "n", + "presence_penalty", + "seed", + "stop", + "stream", + "stream_options", + "temperature", + "top_p", + "tools", + "tool_choice", + "function_call", + "functions", + "max_retries", + "extra_headers", + "parallel_tool_calls", + "audio", + "web_search_options", + "service_tier", + "safety_identifier", + "prompt_cache_key", + "prompt_cache_retention", + "store", + "user", # standard OpenAI Chat Completions API param; passed for all OpenAI-compatible endpoints + ] # works across all models + + model_specific_params = [] + if ( + model != "gpt-3.5-turbo-16k" and model != "gpt-4" + ): # gpt-4 does not support 'response_format' + model_specific_params.append("response_format") + + # Normalize model name for responses API (e.g., "responses/gpt-4.1" -> "gpt-4.1") + model_for_check = ( + model.split("responses/", 1)[1] if "responses/" in model else model + ) + _ = model_for_check # retained for future model-specific logic + return base_params + model_specific_params + + def _map_openai_params( + self, + non_default_params: dict, + optional_params: dict, + model: str, + drop_params: bool, + ) -> dict: + """ + If any supported_openai_params are in non_default_params, add them to optional_params, so they are use in API call + + Args: + non_default_params (dict): Non-default parameters to filter. + optional_params (dict): Optional parameters to update. + model (str): Model name for parameter support check. + + Returns: + dict: Updated optional_params with supported non-default parameters. + """ + supported_openai_params = self.get_supported_openai_params(model) + for param, value in non_default_params.items(): + if param in supported_openai_params: + optional_params[param] = value + return optional_params + + def map_openai_params( + self, + non_default_params: dict, + optional_params: dict, + model: str, + drop_params: bool, + ) -> dict: + return self._map_openai_params( + non_default_params=non_default_params, + optional_params=optional_params, + model=model, + drop_params=drop_params, + ) + + def contains_pdf_url(self, content_item: ChatCompletionFileObjectFile) -> bool: + potential_pdf_url_starts = ["https://", "http://", "www."] + file_id = content_item.get("file_id") + if file_id and any( + file_id.startswith(start) for start in potential_pdf_url_starts + ): + return True + return False + + def _handle_pdf_url( + self, content_item: ChatCompletionFileObjectFile + ) -> ChatCompletionFileObjectFile: + content_copy = content_item.copy() + file_id = content_copy.get("file_id") + if file_id is not None: + base64_data = convert_url_to_base64(file_id) + content_copy["file_data"] = base64_data + content_copy["filename"] = "my_file.pdf" + content_copy.pop("file_id") + return content_copy + + async def _async_handle_pdf_url( + self, content_item: ChatCompletionFileObjectFile + ) -> ChatCompletionFileObjectFile: + file_id = content_item.get("file_id") + if file_id is not None: # check for file id being url done in _handle_pdf_url + base64_data = await async_convert_url_to_base64(file_id) + content_item["file_data"] = base64_data + content_item["filename"] = "my_file.pdf" + content_item.pop("file_id") + return content_item + + def _common_file_data_check( + self, content_item: ChatCompletionFileObjectFile + ) -> ChatCompletionFileObjectFile: + file_data = content_item.get("file_data") + filename = content_item.get("filename") + if file_data is not None and filename is None: + content_item["filename"] = "my_file.pdf" + return content_item + + def _apply_common_transform_content_item( + self, + content_item: OpenAIMessageContentListBlock, + ) -> OpenAIMessageContentListBlock: + litellm_specific_params = {"format"} + if content_item.get("type") == "image_url": + content_item = cast(ChatCompletionImageObject, content_item) + if isinstance(content_item["image_url"], str): + content_item["image_url"] = { + "url": content_item["image_url"], + } + elif isinstance(content_item["image_url"], dict): + new_image_url_obj = ChatCompletionImageUrlObject( + **{ # type: ignore + k: v + for k, v in content_item["image_url"].items() + if k not in litellm_specific_params + } + ) + content_item["image_url"] = new_image_url_obj + elif content_item.get("type") == "file": + content_item = cast(ChatCompletionFileObject, content_item) + file_obj = content_item["file"] + new_file_obj = ChatCompletionFileObjectFile( + **{ # type: ignore + k: v + for k, v in file_obj.items() + if k not in litellm_specific_params + } + ) + content_item["file"] = new_file_obj + + return content_item + + def _transform_content_item( + self, + content_item: OpenAIMessageContentListBlock, + ) -> OpenAIMessageContentListBlock: + content_item = self._apply_common_transform_content_item(content_item) + content_item_type = content_item.get("type") + potential_file_obj = content_item.get("file") + if content_item_type == "file" and potential_file_obj: + file_obj = cast(ChatCompletionFileObjectFile, potential_file_obj) + content_item_typed = cast(ChatCompletionFileObject, content_item) + if self.contains_pdf_url(file_obj): + file_obj = self._handle_pdf_url(file_obj) + file_obj = self._common_file_data_check(file_obj) + content_item_typed["file"] = file_obj + content_item = content_item_typed + return content_item + + async def _async_transform_content_item( + self, content_item: OpenAIMessageContentListBlock, is_async: bool = False + ) -> OpenAIMessageContentListBlock: + content_item = self._apply_common_transform_content_item(content_item) + content_item_type = content_item.get("type") + potential_file_obj = content_item.get("file") + if content_item_type == "file" and potential_file_obj: + file_obj = cast(ChatCompletionFileObjectFile, potential_file_obj) + content_item_typed = cast(ChatCompletionFileObject, content_item) + if self.contains_pdf_url(file_obj): + file_obj = await self._async_handle_pdf_url(file_obj) + file_obj = self._common_file_data_check(file_obj) + content_item_typed["file"] = file_obj + content_item = content_item_typed + return content_item + + # fmt: off + + @overload + def _transform_messages( + self, messages: List[AllMessageValues], model: str, is_async: Literal[True] + ) -> Coroutine[Any, Any, List[AllMessageValues]]: + ... + + @overload + def _transform_messages( + self, + messages: List[AllMessageValues], + model: str, + is_async: Literal[False] = False, + ) -> List[AllMessageValues]: + ... + + # fmt: on + + def _transform_messages( + self, messages: List[AllMessageValues], model: str, is_async: bool = False + ) -> Union[List[AllMessageValues], Coroutine[Any, Any, List[AllMessageValues]]]: + """OpenAI no longer supports image_url as a string, so we need to convert it to a dict""" + + async def _async_transform(): + for message in messages: + message_content = message.get("content") + message_role = message.get("role") + + if ( + message_role == "user" + and message_content + and isinstance(message_content, list) + ): + message_content_types = cast( + List[OpenAIMessageContentListBlock], message_content + ) + for i, content_item in enumerate(message_content_types): + message_content_types[ + i + ] = await self._async_transform_content_item( + cast(OpenAIMessageContentListBlock, content_item), + ) + return messages + + if is_async: + return _async_transform() + else: + for message in messages: + message_content = message.get("content") + message_role = message.get("role") + if ( + message_role == "user" + and message_content + and isinstance(message_content, list) + ): + message_content_types = cast( + List[OpenAIMessageContentListBlock], message_content + ) + for i, content_item in enumerate(message_content): + message_content_types[i] = self._transform_content_item( + cast(OpenAIMessageContentListBlock, content_item) + ) + return messages + + def remove_cache_control_flag_from_messages_and_tools( + self, + model: str, # allows overrides to selectively run this + messages: List[AllMessageValues], + tools: Optional[List["ChatCompletionToolParam"]] = None, + ) -> Tuple[List[AllMessageValues], Optional[List["ChatCompletionToolParam"]]]: + from litellm.litellm_core_utils.prompt_templates.common_utils import ( + filter_value_from_dict, + ) + from litellm.types.llms.openai import ChatCompletionToolParam + + for i, message in enumerate(messages): + messages[i] = cast( + AllMessageValues, filter_value_from_dict(message, "cache_control") # type: ignore + ) + if tools is not None: + for i, tool in enumerate(tools): + tools[i] = cast( + ChatCompletionToolParam, + filter_value_from_dict(tool, "cache_control"), # type: ignore + ) + return messages, tools + + def transform_request( + self, + model: str, + messages: List[AllMessageValues], + optional_params: dict, + litellm_params: dict, + headers: dict, + ) -> dict: + """ + Transform the overall request to be sent to the API. + + Returns: + dict: The transformed request. Sent as the body of the API call. + """ + messages = self._transform_messages(messages=messages, model=model) + messages, tools = self.remove_cache_control_flag_from_messages_and_tools( + model=model, messages=messages, tools=optional_params.get("tools", []) + ) + if tools is not None and len(tools) > 0: + optional_params["tools"] = tools + + optional_params.pop("max_retries", None) + + return { + "model": model, + "messages": messages, + **optional_params, + } + + async def async_transform_request( + self, + model: str, + messages: List[AllMessageValues], + optional_params: dict, + litellm_params: dict, + headers: dict, + ) -> dict: + transformed_messages = await self._transform_messages( + messages=messages, model=model, is_async=True + ) + ( + transformed_messages, + tools, + ) = self.remove_cache_control_flag_from_messages_and_tools( + model=model, + messages=transformed_messages, + tools=optional_params.get("tools", []), + ) + if tools is not None and len(tools) > 0: + optional_params["tools"] = tools + if self.__class__._is_base_class: + return { + "model": model, + "messages": transformed_messages, + **optional_params, + } + else: + ## allow for any object specific behaviour to be handled + return self.transform_request( + model, messages, optional_params, litellm_params, headers + ) + + def _passed_in_tools(self, optional_params: dict) -> bool: + return optional_params.get("tools", None) is not None + + def _check_and_fix_if_content_is_tool_call( + self, content: str, optional_params: dict + ) -> Optional[ChatCompletionMessageToolCall]: + """ + Check if the content is a tool call + """ + import json + + if not self._passed_in_tools(optional_params): + return None + tool_call_names = get_tool_call_names(optional_params.get("tools", [])) + try: + json_content = json.loads(content) + if ( + json_content.get("type") == "function" + and json_content.get("name") in tool_call_names + ): + return ChatCompletionMessageToolCall( + function=Function( + name=json_content.get("name"), + arguments=json_content.get("arguments"), + ) + ) + except Exception: + return None + + return None + + def _get_finish_reason(self, message: Message, received_finish_reason: str) -> str: + if message.tool_calls is not None: + return "tool_calls" + else: + return received_finish_reason + + def _transform_choices( + self, + choices: List[OpenAIChatCompletionChoices], + json_mode: Optional[bool] = None, + optional_params: Optional[dict] = None, + ) -> List[Choices]: + transformed_choices = [] + + for choice in choices: + ## HANDLE JSON MODE - anthropic returns single function call] + tool_calls = choice["message"].get("tool_calls", None) + new_tool_calls: Optional[List[ChatCompletionMessageToolCall]] = None + message_content = choice["message"].get("content", None) + if tool_calls is not None: + _openai_tool_calls = [] + for _tc in tool_calls: + _openai_tc = ChatCompletionMessageToolCall(**_tc) # type: ignore + _openai_tool_calls.append(_openai_tc) + fixed_tool_calls = _handle_invalid_parallel_tool_calls( + _openai_tool_calls + ) + + if fixed_tool_calls is not None: + new_tool_calls = fixed_tool_calls + elif ( + optional_params is not None + and message_content + and isinstance(message_content, str) + ): + new_tool_call = self._check_and_fix_if_content_is_tool_call( + message_content, optional_params + ) + if new_tool_call is not None: + choice["message"]["content"] = None # remove the content + new_tool_calls = [new_tool_call] + + translated_message: Optional[Message] = None + finish_reason: Optional[str] = None + if new_tool_calls and _should_convert_tool_call_to_json_mode( + tool_calls=new_tool_calls, + convert_tool_call_to_json_mode=json_mode, + ): + # to support response_format on claude models + json_mode_content_str: Optional[str] = ( + str(new_tool_calls[0]["function"].get("arguments", "")) or None + ) + if json_mode_content_str is not None: + translated_message = Message(content=json_mode_content_str) + finish_reason = "stop" + + if translated_message is None: + ## get the reasoning content + ( + reasoning_content, + content_str, + ) = _extract_reasoning_content(cast(dict, choice["message"])) + + translated_message = Message( + role="assistant", + content=content_str, + reasoning_content=reasoning_content, + thinking_blocks=None, + tool_calls=new_tool_calls, + ) + + if finish_reason is None: + finish_reason = choice["finish_reason"] + + translated_choice = Choices( + finish_reason=finish_reason, + index=choice["index"], + message=translated_message, + logprobs=None, + enhancements=None, + ) + + translated_choice.finish_reason = map_finish_reason( + self._get_finish_reason(translated_message, choice["finish_reason"]) + ) + transformed_choices.append(translated_choice) + + return transformed_choices + + def transform_response( + self, + model: str, + raw_response: httpx.Response, + model_response: ModelResponse, + logging_obj: LiteLLMLoggingObj, + request_data: dict, + messages: List[AllMessageValues], + optional_params: dict, + litellm_params: dict, + encoding: Any, + api_key: Optional[str] = None, + json_mode: Optional[bool] = None, + ) -> ModelResponse: + """ + Transform the response from the API. + + Returns: + dict: The transformed response. + """ + + ## LOGGING + logging_obj.post_call( + input=messages, + api_key=api_key, + original_response=raw_response.text, + additional_args={"complete_input_dict": request_data}, + ) + + ## RESPONSE OBJECT + try: + completion_response = raw_response.json() + except Exception as e: + response_headers = getattr(raw_response, "headers", None) + raise OpenAIError( + message="Unable to get json response - {}, Original Response: {}".format( + str(e), raw_response.text + ), + status_code=raw_response.status_code, + headers=response_headers, + ) + raw_response_headers = dict(raw_response.headers) + final_response_obj = convert_to_model_response_object( + response_object=completion_response, + model_response_object=model_response, + hidden_params={"headers": raw_response_headers}, + _response_headers=raw_response_headers, + ) + + return cast(ModelResponse, final_response_obj) + + def get_error_class( + self, error_message: str, status_code: int, headers: Union[dict, httpx.Headers] + ) -> BaseLLMException: + return OpenAIError( + status_code=status_code, + message=error_message, + headers=cast(httpx.Headers, headers), + ) + + def get_complete_url( + self, + api_base: Optional[str], + api_key: Optional[str], + model: str, + optional_params: dict, + litellm_params: dict, + stream: Optional[bool] = None, + ) -> str: + """ + Get the complete URL for the API call. + + Returns: + str: The complete URL for the API call. + """ + if api_base is None: + api_base = "https://api.openai.com" + endpoint = "chat/completions" + + # Remove trailing slash from api_base if present + api_base = api_base.rstrip("/") + + # Check if endpoint is already in the api_base + if endpoint in api_base: + return api_base + + return f"{api_base}/{endpoint}" + + def validate_environment( + self, + headers: dict, + model: str, + messages: List[AllMessageValues], + optional_params: dict, + litellm_params: dict, + api_key: Optional[str] = None, + api_base: Optional[str] = None, + ) -> dict: + if api_key is not None: + headers["Authorization"] = f"Bearer {api_key}" + + # Ensure Content-Type is set to application/json + if "content-type" not in headers and "Content-Type" not in headers: + headers["Content-Type"] = "application/json" + + return headers + + def get_models( + self, api_key: Optional[str] = None, api_base: Optional[str] = None + ) -> List[str]: + """ + Calls OpenAI's `/v1/models` endpoint and returns the list of models. + """ + + if api_base is None: + api_base = "https://api.openai.com" + if api_key is None: + api_key = get_secret_str("OPENAI_API_KEY") + + # Strip api_base to just the base URL (scheme + host + port) + parsed_url = httpx.URL(api_base) + base_url = f"{parsed_url.scheme}://{parsed_url.host}" + if parsed_url.port: + base_url += f":{parsed_url.port}" + + response = litellm.module_level_client.get( + url=f"{base_url}/v1/models", + headers={"Authorization": f"Bearer {api_key}"}, + ) + + if response.status_code != 200: + raise Exception(f"Failed to get models: {response.text}") + + models = response.json()["data"] + return [model["id"] for model in models] + + @staticmethod + def get_api_key(api_key: Optional[str] = None) -> Optional[str]: + return ( + api_key + or litellm.api_key + or litellm.openai_key + or get_secret_str("OPENAI_API_KEY") + ) + + @staticmethod + def get_api_base(api_base: Optional[str] = None) -> Optional[str]: + return ( + api_base + or litellm.api_base + or get_secret_str("OPENAI_BASE_URL") + or get_secret_str("OPENAI_API_BASE") + or "https://api.openai.com/v1" + ) + + @staticmethod + def get_base_model(model: Optional[str] = None) -> Optional[str]: + return model + + def get_token_counter(self) -> Optional["BaseTokenCounter"]: + from litellm.llms.openai.responses.count_tokens.token_counter import ( + OpenAITokenCounter, + ) + + return OpenAITokenCounter() + + def get_model_response_iterator( + self, + streaming_response: Union[Iterator[str], AsyncIterator[str], ModelResponse], + sync_stream: bool, + json_mode: Optional[bool] = False, + ) -> Any: + return OpenAIChatCompletionStreamingHandler( + streaming_response=streaming_response, + sync_stream=sync_stream, + json_mode=json_mode, + ) + + +class OpenAIChatCompletionStreamingHandler(BaseModelResponseIterator): + def _map_reasoning_to_reasoning_content(self, choices: list) -> list: + """ + Map 'reasoning' field to 'reasoning_content' field in delta. + + Some OpenAI-compatible providers (e.g., GLM-5, hosted_vllm) return + delta.reasoning, but LiteLLM expects delta.reasoning_content. + + Args: + choices: List of choice objects from the streaming chunk + + Returns: + List of choices with reasoning field mapped to reasoning_content + """ + for choice in choices: + delta = choice.get("delta", {}) + if "reasoning" in delta: + delta["reasoning_content"] = delta.pop("reasoning") + return choices + + def chunk_parser(self, chunk: dict) -> ModelResponseStream: + try: + choices = chunk.get("choices", []) + choices = self._map_reasoning_to_reasoning_content(choices) + + kwargs: Dict[str, Any] = { + "id": chunk.get("id"), + "object": "chat.completion.chunk", + "created": chunk.get("created"), + "model": chunk.get("model"), + "choices": choices, + } + if "usage" in chunk and chunk["usage"] is not None: + kwargs["usage"] = chunk["usage"] + return ModelResponseStream(**kwargs) + except Exception as e: + raise e