diff --git a/litellm/completion_extras/litellm_responses_transformation/transformation.py b/litellm/completion_extras/litellm_responses_transformation/transformation.py index ee4cdbcdf36..d1b36c9b6d7 100644 --- a/litellm/completion_extras/litellm_responses_transformation/transformation.py +++ b/litellm/completion_extras/litellm_responses_transformation/transformation.py @@ -27,6 +27,10 @@ import litellm from litellm import ModelResponse from litellm._logging import verbose_logger from litellm.llms.base_llm.base_model_iterator import BaseModelResponseIterator +from litellm.responses.format_mapping import ( + normalize_provider_specific_fields, + response_format_to_text_format, +) from litellm.llms.base_llm.bridges.completion_transformation import ( CompletionTransformationBridge, ) @@ -106,16 +110,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): # Handle function_call items (e.g., from GPT-5 Codex format) if item_type == "function_call": - # Extract provider_specific_fields if present and pass through as-is - provider_specific_fields = item.get("provider_specific_fields") - if provider_specific_fields and not isinstance( - provider_specific_fields, dict - ): - provider_specific_fields = ( - dict(provider_specific_fields) - if hasattr(provider_specific_fields, "__dict__") - else {} - ) + provider_specific_fields = normalize_provider_specific_fields(item) tool_call_dict = { "id": item.get("call_id") or item.get("id", ""), @@ -875,51 +870,8 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): def _transform_response_format_to_text_format( self, response_format: Union[Dict[str, Any], Any] ) -> Optional[Dict[str, Any]]: - """ - Transform Chat Completion response_format parameter to Responses API text.format parameter. - - Chat Completion response_format structure: - { - "type": "json_schema", - "json_schema": { - "name": "schema_name", - "schema": {...}, - "strict": True - } - } - - Responses API text parameter structure: - { - "format": { - "type": "json_schema", - "name": "schema_name", - "schema": {...}, - "strict": True - } - } - """ - if not response_format: - return None - - if isinstance(response_format, dict): - format_type = response_format.get("type") - - if format_type == "json_schema": - json_schema = response_format.get("json_schema", {}) - return { - "format": { - "type": "json_schema", - "name": json_schema.get("name", "response_schema"), - "schema": json_schema.get("schema", {}), - "strict": json_schema.get("strict", False), - } - } - elif format_type == "json_object": - return {"format": {"type": "json_object"}} - elif format_type == "text": - return {"format": {"type": "text"}} - - return None + """Transform Chat Completion response_format to Responses API text.format.""" + return response_format_to_text_format(response_format) @staticmethod def _convert_annotations_to_chat_format( @@ -961,21 +913,6 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): return result if result else None - def _map_responses_status_to_finish_reason(self, status: Optional[str]) -> str: - """Map responses API status to chat completion finish_reason""" - if not status: - return "stop" - - status_mapping = { - "completed": "stop", - "incomplete": "length", - "failed": "stop", - "cancelled": "stop", - } - - return status_mapping.get(status, "stop") - - class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator): def __init__( self, streaming_response, sync_stream: bool, json_mode: Optional[bool] = False @@ -1055,16 +992,7 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator): # New output item added output_item = parsed_chunk.get("item", {}) if output_item.get("type") == "function_call": - # Extract provider_specific_fields if present - provider_specific_fields = output_item.get("provider_specific_fields") - if provider_specific_fields and not isinstance( - provider_specific_fields, dict - ): - provider_specific_fields = ( - dict(provider_specific_fields) - if hasattr(provider_specific_fields, "__dict__") - else {} - ) + provider_specific_fields = normalize_provider_specific_fields(output_item) function_chunk = ChatCompletionToolCallFunctionChunk( name=output_item.get("name", None), @@ -1129,23 +1057,13 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator): # New output item added output_item = parsed_chunk.get("item", {}) if output_item.get("type") == "function_call": - # Extract provider_specific_fields if present - provider_specific_fields = output_item.get("provider_specific_fields") - if provider_specific_fields and not isinstance( - provider_specific_fields, dict - ): - provider_specific_fields = ( - dict(provider_specific_fields) - if hasattr(provider_specific_fields, "__dict__") - else {} - ) + provider_specific_fields = normalize_provider_specific_fields(output_item) function_chunk = ChatCompletionToolCallFunctionChunk( name=output_item.get("name", None), arguments="", # responses API sends everything again, we don't ) - # Add provider_specific_fields to function if present if provider_specific_fields: function_chunk[ "provider_specific_fields" diff --git a/litellm/responses/format_mapping.py b/litellm/responses/format_mapping.py new file mode 100644 index 00000000000..a26d462eea3 --- /dev/null +++ b/litellm/responses/format_mapping.py @@ -0,0 +1,342 @@ +""" +Shared bidirectional mapping dictionaries between Responses API and Chat Completions formats. + +Both bridges (Responses→CC and CC→Responses) import from here so the +mapping knowledge lives in one place. Asymmetries are documented inline. +""" + +from typing import Any, Dict, Optional, Union + +from litellm.types.llms.openai import ( + InputTokensDetails, + OutputTokensDetails, + ResponseAPIUsage, + ResponsesAPIStatus, +) +from litellm.types.utils import ( + CompletionTokensDetailsWrapper, + ModelResponse, + PromptTokensDetailsWrapper, + Usage, +) + +# --------------------------------------------------------------------------- +# Status (Responses API) <-> finish_reason (Chat Completions) +# --------------------------------------------------------------------------- +# These are NOT perfect inverses. The Responses API status space is larger +# than Chat Completions finish_reason: +# - "failed" and "cancelled" collapse to "stop" (no CC equivalent) +# - "content_filter" maps to "incomplete" (best approximation) +# - "tool_calls" and "function_call" map to "completed" (tools are +# signalled separately in Responses API, not via status) +# --------------------------------------------------------------------------- + +STATUS_TO_FINISH_REASON: dict = { + "completed": "stop", + "incomplete": "length", + "failed": "stop", + "cancelled": "stop", +} + +FINISH_REASON_TO_STATUS: dict = { + "stop": "completed", + "tool_calls": "completed", + "function_call": "completed", + "length": "incomplete", + "content_filter": "incomplete", +} + + +# --------------------------------------------------------------------------- +# Provider-specific fields extraction +# --------------------------------------------------------------------------- +# Both bridges repeatedly normalize provider_specific_fields from objects or +# dicts into a plain dict. This helper consolidates that boilerplate. +# --------------------------------------------------------------------------- + + +def normalize_provider_specific_fields(obj: Any) -> Optional[Dict[str, Any]]: + """Extract and normalize provider_specific_fields from a dict or object to a plain dict.""" + if isinstance(obj, dict): + psf = obj.get("provider_specific_fields") + else: + psf = getattr(obj, "provider_specific_fields", None) + + if not psf: + return None + if isinstance(psf, dict): + return psf + return dict(psf) if hasattr(psf, "__dict__") else {} + + +def status_to_finish_reason(status: Optional[str]) -> str: + """Map Responses API status to Chat Completions finish_reason.""" + if not status: + return "stop" + return STATUS_TO_FINISH_REASON.get(status, "stop") + + +def finish_reason_to_status(finish_reason: Optional[str]) -> ResponsesAPIStatus: + """Map Chat Completions finish_reason to Responses API status.""" + if finish_reason is None: + return "completed" + return FINISH_REASON_TO_STATUS.get(finish_reason, "completed") + + +# --------------------------------------------------------------------------- +# response_format (Chat Completions) <-> text.format (Responses API) +# --------------------------------------------------------------------------- +# Chat Completions nests json_schema under a "json_schema" key: +# {"type": "json_schema", "json_schema": {"name": ..., "schema": ..., "strict": ...}} +# +# Responses API nests everything under "format": +# {"format": {"type": "json_schema", "name": ..., "schema": ..., "strict": ...}} +# +# Asymmetry: "text" type produces {"format": {"type": "text"}} in the +# Responses direction, but returns None in the CC direction (it's the +# implicit default in Chat Completions). +# --------------------------------------------------------------------------- + + +def response_format_to_text_format( + response_format: Union[Dict[str, Any], Any], +) -> Optional[Dict[str, Any]]: + """ + Chat Completion response_format -> Responses API text param. + + Returns dict with "format" key, or None. + """ + if not response_format: + return None + + if isinstance(response_format, dict): + format_type = response_format.get("type") + + if format_type == "json_schema": + json_schema = response_format.get("json_schema", {}) + return { + "format": { + "type": "json_schema", + "name": json_schema.get("name", "response_schema"), + "schema": json_schema.get("schema", {}), + "strict": json_schema.get("strict", False), + } + } + elif format_type == "json_object": + return {"format": {"type": "json_object"}} + elif format_type == "text": + return {"format": {"type": "text"}} + + return None + + +def text_format_to_response_format( + text_param: Union[Dict[str, Any], Any], +) -> Optional[Dict[str, Any]]: + """ + Responses API text param -> Chat Completion response_format. + + Returns None for "text" type (implicit default in CC). + """ + if not text_param: + return None + + if isinstance(text_param, dict): + format_param = text_param.get("format") + if format_param and isinstance(format_param, dict): + format_type = format_param.get("type") + + if format_type == "json_schema": + return { + "type": "json_schema", + "json_schema": { + "name": format_param.get("name", "response_schema"), + "schema": format_param.get("schema", {}), + "strict": format_param.get("strict", False), + }, + } + elif format_type == "json_object": + return {"type": "json_object"} + elif format_type == "text": + return None + + return None + + +# --------------------------------------------------------------------------- +# Usage field mapping (Chat Completions <-> Responses API) +# --------------------------------------------------------------------------- +# Chat Completions uses prompt_tokens/completion_tokens with +# prompt_tokens_details/completion_tokens_details. +# Responses API uses input_tokens/output_tokens with +# input_tokens_details/output_tokens_details. +# +# The field names differ but the structure is the same. Both directions +# also preserve the "cost" attribute if present. +# +# Asymmetry: the CC→Responses direction defaults cached_tokens and +# reasoning_tokens to 0 when absent; the Responses→CC direction passes +# None through. This is pre-existing behavior preserved as-is. +# --------------------------------------------------------------------------- + + +def response_api_usage_to_chat_usage( + usage_input: Optional[Union[dict, ResponseAPIUsage]], +) -> Usage: + """Transform ResponseAPIUsage to Chat Completions Usage.""" + if usage_input is None: + return Usage(prompt_tokens=0, completion_tokens=0, total_tokens=0) + + response_api_usage: ResponseAPIUsage + if isinstance(usage_input, dict): + total_tokens = usage_input.get("total_tokens") + if total_tokens is None: + input_tokens = usage_input.get("input_tokens") + output_tokens = usage_input.get("output_tokens") + if input_tokens is not None and output_tokens is not None: + total_tokens = input_tokens + output_tokens + usage_input["total_tokens"] = total_tokens + response_api_usage = ResponseAPIUsage(**usage_input) + else: + response_api_usage = usage_input + + prompt_tokens: int = response_api_usage.input_tokens or 0 + completion_tokens: int = response_api_usage.output_tokens or 0 + + prompt_tokens_details: Optional[PromptTokensDetailsWrapper] = None + if response_api_usage.input_tokens_details: + if isinstance(response_api_usage.input_tokens_details, dict): + prompt_tokens_details = PromptTokensDetailsWrapper( + **response_api_usage.input_tokens_details + ) + else: + prompt_tokens_details = PromptTokensDetailsWrapper( + cached_tokens=getattr( + response_api_usage.input_tokens_details, "cached_tokens", None + ), + audio_tokens=getattr( + response_api_usage.input_tokens_details, "audio_tokens", None + ), + text_tokens=getattr( + response_api_usage.input_tokens_details, "text_tokens", None + ), + image_tokens=getattr( + response_api_usage.input_tokens_details, "image_tokens", None + ), + ) + + completion_tokens_details: Optional[CompletionTokensDetailsWrapper] = None + output_tokens_details = getattr( + response_api_usage, "output_tokens_details", None + ) + if output_tokens_details: + completion_tokens_details = CompletionTokensDetailsWrapper( + reasoning_tokens=getattr( + output_tokens_details, "reasoning_tokens", None + ), + image_tokens=getattr(output_tokens_details, "image_tokens", None), + text_tokens=getattr(output_tokens_details, "text_tokens", None), + ) + + chat_usage = Usage( + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + total_tokens=prompt_tokens + completion_tokens, + prompt_tokens_details=prompt_tokens_details, + completion_tokens_details=completion_tokens_details, + ) + + if hasattr(response_api_usage, "cost") and response_api_usage.cost is not None: + setattr(chat_usage, "cost", response_api_usage.cost) + + return chat_usage + + +def chat_usage_to_response_api_usage( + chat_completion_response: Union[ModelResponse, Usage], +) -> ResponseAPIUsage: + """Transform Chat Completions Usage to ResponseAPIUsage.""" + if isinstance(chat_completion_response, ModelResponse): + usage: Optional[Usage] = getattr(chat_completion_response, "usage", None) + else: + usage = chat_completion_response + + if usage is None: + return ResponseAPIUsage(input_tokens=0, output_tokens=0, total_tokens=0) + + response_usage = ResponseAPIUsage( + input_tokens=usage.prompt_tokens, + output_tokens=usage.completion_tokens, + total_tokens=usage.total_tokens, + ) + + if hasattr(usage, "cost") and usage.cost is not None: + setattr(response_usage, "cost", usage.cost) + + # Translate prompt_tokens_details to input_tokens_details + if ( + hasattr(usage, "prompt_tokens_details") + and usage.prompt_tokens_details is not None + ): + prompt_details = usage.prompt_tokens_details + input_details_dict: Dict[str, int] = {} + + if ( + hasattr(prompt_details, "cached_tokens") + and prompt_details.cached_tokens is not None + ): + input_details_dict["cached_tokens"] = prompt_details.cached_tokens + else: + input_details_dict["cached_tokens"] = 0 + + if ( + hasattr(prompt_details, "text_tokens") + and prompt_details.text_tokens is not None + ): + input_details_dict["text_tokens"] = prompt_details.text_tokens + + if ( + hasattr(prompt_details, "audio_tokens") + and prompt_details.audio_tokens is not None + ): + input_details_dict["audio_tokens"] = prompt_details.audio_tokens + + if input_details_dict: + response_usage.input_tokens_details = InputTokensDetails( + **input_details_dict + ) + + # Translate completion_tokens_details to output_tokens_details + if ( + hasattr(usage, "completion_tokens_details") + and usage.completion_tokens_details is not None + ): + completion_details = usage.completion_tokens_details + output_details_dict: Dict[str, int] = {} + + if ( + hasattr(completion_details, "reasoning_tokens") + and completion_details.reasoning_tokens is not None + ): + output_details_dict["reasoning_tokens"] = completion_details.reasoning_tokens + else: + output_details_dict["reasoning_tokens"] = 0 + + if ( + hasattr(completion_details, "text_tokens") + and completion_details.text_tokens is not None + ): + output_details_dict["text_tokens"] = completion_details.text_tokens + + if ( + hasattr(completion_details, "image_tokens") + and completion_details.image_tokens is not None + ): + output_details_dict["image_tokens"] = completion_details.image_tokens + + if output_details_dict: + response_usage.output_tokens_details = OutputTokensDetails( + **output_details_dict + ) + + return response_usage diff --git a/litellm/responses/litellm_completion_transformation/transformation.py b/litellm/responses/litellm_completion_transformation/transformation.py index b6479a36998..f172b6a7a2b 100644 --- a/litellm/responses/litellm_completion_transformation/transformation.py +++ b/litellm/responses/litellm_completion_transformation/transformation.py @@ -15,6 +15,11 @@ from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging from litellm.responses.litellm_completion_transformation.session_handler import ( ResponsesSessionHandler, ) +from litellm.responses.format_mapping import ( + finish_reason_to_status, + normalize_provider_specific_fields, + text_format_to_response_format, +) from litellm.types.llms.openai import ( AllMessageValues, ChatCompletionImageObject, @@ -1462,29 +1467,7 @@ class LiteLLMCompletionResponsesConfig: for tool in all_chat_completion_tools: if tool.type == "function": function_definition = tool.function - provider_specific_fields: Optional[Dict] = None - if hasattr(tool, "provider_specific_fields") and getattr( - tool, "provider_specific_fields", None - ): - provider_specific_fields = getattr(tool, "provider_specific_fields") - if not isinstance(provider_specific_fields, dict): - provider_specific_fields = ( - dict(provider_specific_fields) # type: ignore - if hasattr(provider_specific_fields, "__dict__") - else {} - ) - elif hasattr( - function_definition, "provider_specific_fields" - ) and getattr(function_definition, "provider_specific_fields", None): - provider_specific_fields = getattr( - function_definition, "provider_specific_fields" - ) - if not isinstance(provider_specific_fields, dict): - provider_specific_fields = ( - dict(provider_specific_fields) # type: ignore - if hasattr(provider_specific_fields, "__dict__") - else {} - ) + provider_specific_fields = normalize_provider_specific_fields(tool) or normalize_provider_specific_fields(function_definition) output_tool_call: ResponseFunctionToolCall = ResponseFunctionToolCall( name=function_definition.name or "", @@ -1506,29 +1489,8 @@ class LiteLLMCompletionResponsesConfig: def _map_chat_completion_finish_reason_to_responses_status( finish_reason: Optional[str], ) -> ResponsesAPIStatus: - """ - Map chat completion finish_reason to responses API status. - - Chat completion finish_reason values include: "stop", "length", "tool_calls", "content_filter", "function_call" - Responses API status values are: "completed", "failed", "in_progress", "cancelled", "queued", "incomplete" - - Args: - finish_reason: The finish_reason from a chat completion response - - Returns: - The corresponding responses API status value (one of ResponsesAPIStatus) - """ - if finish_reason is None: - return "completed" - - # Map finish reasons to status - if finish_reason in ["stop", "tool_calls", "function_call"]: - return "completed" - elif finish_reason in ["length", "content_filter"]: - return "incomplete" - else: - # Default to completed for unknown finish reasons - return "completed" + """Map chat completion finish_reason to responses API status.""" + return finish_reason_to_status(finish_reason) @staticmethod def convert_response_function_tool_call_to_chat_completion_tool_call( @@ -1545,28 +1507,7 @@ class LiteLLMCompletionResponsesConfig: Returns: Dictionary in ChatCompletionToolCallChunk format """ - # Extract provider_specific_fields if present - provider_specific_fields = getattr( - tool_call_item, "provider_specific_fields", None - ) - if provider_specific_fields and not isinstance(provider_specific_fields, dict): - provider_specific_fields = ( - dict(provider_specific_fields) - if hasattr(provider_specific_fields, "__dict__") - else {} - ) - elif hasattr(tool_call_item, "get") and callable(tool_call_item.get): # type: ignore - provider_fields = tool_call_item.get("provider_specific_fields") # type: ignore - if provider_fields: - provider_specific_fields = ( - provider_fields - if isinstance(provider_fields, dict) - else ( - dict(provider_fields) # type: ignore - if hasattr(provider_fields, "__dict__") - else {} - ) - ) + provider_specific_fields = normalize_provider_specific_fields(tool_call_item) function_dict: Dict[str, Any] = { "name": tool_call_item.name, @@ -2048,143 +1989,14 @@ class LiteLLMCompletionResponsesConfig: def _transform_chat_completion_usage_to_responses_usage( chat_completion_response: Union[ModelResponse, Usage], ) -> ResponseAPIUsage: - if isinstance(chat_completion_response, ModelResponse): - usage: Optional[Usage] = getattr(chat_completion_response, "usage", None) - else: - usage = chat_completion_response - if usage is None: - return ResponseAPIUsage( - input_tokens=0, - output_tokens=0, - total_tokens=0, - ) + """Transform Chat Completions Usage to ResponseAPIUsage.""" + from litellm.responses.format_mapping import chat_usage_to_response_api_usage - response_usage = ResponseAPIUsage( - input_tokens=usage.prompt_tokens, - output_tokens=usage.completion_tokens, - total_tokens=usage.total_tokens, - ) - - # Preserve cost field if it exists (for streaming usage with cost calculation) - if hasattr(usage, "cost") and usage.cost is not None: - setattr(response_usage, "cost", usage.cost) - - # Translate prompt_tokens_details to input_tokens_details - if ( - hasattr(usage, "prompt_tokens_details") - and usage.prompt_tokens_details is not None - ): - prompt_details = usage.prompt_tokens_details - input_details_dict: Dict[str, int] = {} - - if ( - hasattr(prompt_details, "cached_tokens") - and prompt_details.cached_tokens is not None - ): - input_details_dict["cached_tokens"] = prompt_details.cached_tokens - else: - input_details_dict["cached_tokens"] = 0 - - if ( - hasattr(prompt_details, "text_tokens") - and prompt_details.text_tokens is not None - ): - input_details_dict["text_tokens"] = prompt_details.text_tokens - - if ( - hasattr(prompt_details, "audio_tokens") - and prompt_details.audio_tokens is not None - ): - input_details_dict["audio_tokens"] = prompt_details.audio_tokens - - if input_details_dict: - response_usage.input_tokens_details = InputTokensDetails( - **input_details_dict - ) - - # Translate completion_tokens_details to output_tokens_details - if ( - hasattr(usage, "completion_tokens_details") - and usage.completion_tokens_details is not None - ): - completion_details = usage.completion_tokens_details - output_details_dict: Dict[str, int] = {} - if ( - hasattr(completion_details, "reasoning_tokens") - and completion_details.reasoning_tokens is not None - ): - output_details_dict[ - "reasoning_tokens" - ] = completion_details.reasoning_tokens - else: - output_details_dict["reasoning_tokens"] = 0 - - if ( - hasattr(completion_details, "text_tokens") - and completion_details.text_tokens is not None - ): - output_details_dict["text_tokens"] = completion_details.text_tokens - - if ( - hasattr(completion_details, "image_tokens") - and completion_details.image_tokens is not None - ): - output_details_dict["image_tokens"] = completion_details.image_tokens - - if output_details_dict: - response_usage.output_tokens_details = OutputTokensDetails( - **output_details_dict - ) - - return response_usage + return chat_usage_to_response_api_usage(chat_completion_response) @staticmethod def _transform_text_format_to_response_format( text_param: Union[Dict[str, Any], Any], ) -> Optional[Dict[str, Any]]: - """ - Transform Responses API text.format parameter to Chat Completion response_format parameter. - - Responses API text parameter structure: - { - "format": { - "type": "json_schema", - "name": "schema_name", - "schema": {...}, - "strict": True - } - } - - Chat Completion response_format structure: - { - "type": "json_schema", - "json_schema": { - "name": "schema_name", - "schema": {...}, - "strict": True - } - } - """ - if not text_param: - return None - - if isinstance(text_param, dict): - format_param = text_param.get("format") - if format_param and isinstance(format_param, dict): - format_type = format_param.get("type") - - if format_type == "json_schema": - return { - "type": "json_schema", - "json_schema": { - "name": format_param.get("name", "response_schema"), - "schema": format_param.get("schema", {}), - "strict": format_param.get("strict", False), - }, - } - elif format_type == "json_object": - return {"type": "json_object"} - elif format_type == "text": - return None - - return None + """Transform Responses API text.format to Chat Completion response_format.""" + return text_format_to_response_format(text_param) diff --git a/litellm/responses/utils.py b/litellm/responses/utils.py index 11097864225..7c0bd5b6617 100644 --- a/litellm/responses/utils.py +++ b/litellm/responses/utils.py @@ -651,76 +651,7 @@ class ResponseAPILoggingUtils: def _transform_response_api_usage_to_chat_usage( usage_input: Optional[Union[dict, ResponseAPIUsage]], ) -> Usage: - """ - Transforms ResponseAPIUsage or ImageUsage to a Usage object. + """Transform ResponseAPIUsage to Chat Completions Usage.""" + from litellm.responses.format_mapping import response_api_usage_to_chat_usage - Both have the same spec with input_tokens, output_tokens, and - input_tokens_details (text_tokens, image_tokens). - """ - if usage_input is None: - return Usage( - prompt_tokens=0, - completion_tokens=0, - total_tokens=0, - ) - response_api_usage: ResponseAPIUsage - if isinstance(usage_input, dict): - total_tokens = usage_input.get("total_tokens") - if total_tokens is None: - input_tokens = usage_input.get("input_tokens") - output_tokens = usage_input.get("output_tokens") - if input_tokens is not None and output_tokens is not None: - total_tokens = input_tokens + output_tokens - usage_input["total_tokens"] = total_tokens - response_api_usage = ResponseAPIUsage(**usage_input) - else: - response_api_usage = usage_input - prompt_tokens: int = response_api_usage.input_tokens or 0 - completion_tokens: int = response_api_usage.output_tokens or 0 - prompt_tokens_details: Optional[PromptTokensDetailsWrapper] = None - if response_api_usage.input_tokens_details: - if isinstance(response_api_usage.input_tokens_details, dict): - prompt_tokens_details = PromptTokensDetailsWrapper( - **response_api_usage.input_tokens_details - ) - else: - prompt_tokens_details = PromptTokensDetailsWrapper( - cached_tokens=getattr( - response_api_usage.input_tokens_details, "cached_tokens", None - ), - audio_tokens=getattr( - response_api_usage.input_tokens_details, "audio_tokens", None - ), - text_tokens=getattr( - response_api_usage.input_tokens_details, "text_tokens", None - ), - image_tokens=getattr( - response_api_usage.input_tokens_details, "image_tokens", None - ), - ) - completion_tokens_details: Optional[CompletionTokensDetailsWrapper] = None - output_tokens_details = getattr( - response_api_usage, "output_tokens_details", None - ) - if output_tokens_details: - completion_tokens_details = CompletionTokensDetailsWrapper( - reasoning_tokens=getattr( - output_tokens_details, "reasoning_tokens", None - ), - image_tokens=getattr(output_tokens_details, "image_tokens", None), - text_tokens=getattr(output_tokens_details, "text_tokens", None), - ) - - chat_usage = Usage( - prompt_tokens=prompt_tokens, - completion_tokens=completion_tokens, - total_tokens=prompt_tokens + completion_tokens, - prompt_tokens_details=prompt_tokens_details, - completion_tokens_details=completion_tokens_details, - ) - - # Preserve cost attribute if it exists on ResponseAPIUsage - if hasattr(response_api_usage, "cost") and response_api_usage.cost is not None: - setattr(chat_usage, "cost", response_api_usage.cost) - - return chat_usage + return response_api_usage_to_chat_usage(usage_input) diff --git a/tests/test_litellm/responses/test_format_mapping.py b/tests/test_litellm/responses/test_format_mapping.py new file mode 100644 index 00000000000..9e6faba000d --- /dev/null +++ b/tests/test_litellm/responses/test_format_mapping.py @@ -0,0 +1,304 @@ +"""Tests for the shared format mapping between Responses API and Chat Completions.""" + +import pytest + +from litellm.responses.format_mapping import ( + FINISH_REASON_TO_STATUS, + STATUS_TO_FINISH_REASON, + chat_usage_to_response_api_usage, + finish_reason_to_status, + normalize_provider_specific_fields, + response_api_usage_to_chat_usage, + response_format_to_text_format, + status_to_finish_reason, + text_format_to_response_format, +) +from litellm.types.llms.openai import ResponseAPIUsage +from litellm.types.utils import ModelResponse, Usage + + +class TestStatusToFinishReason: + def test_completed(self): + assert status_to_finish_reason("completed") == "stop" + + def test_incomplete(self): + assert status_to_finish_reason("incomplete") == "length" + + def test_failed(self): + assert status_to_finish_reason("failed") == "stop" + + def test_cancelled(self): + assert status_to_finish_reason("cancelled") == "stop" + + def test_none(self): + assert status_to_finish_reason(None) == "stop" + + def test_empty_string(self): + assert status_to_finish_reason("") == "stop" + + def test_unknown(self): + assert status_to_finish_reason("some_future_status") == "stop" + + +class TestFinishReasonToStatus: + def test_stop(self): + assert finish_reason_to_status("stop") == "completed" + + def test_tool_calls(self): + assert finish_reason_to_status("tool_calls") == "completed" + + def test_function_call(self): + assert finish_reason_to_status("function_call") == "completed" + + def test_length(self): + assert finish_reason_to_status("length") == "incomplete" + + def test_content_filter(self): + assert finish_reason_to_status("content_filter") == "incomplete" + + def test_none(self): + assert finish_reason_to_status(None) == "completed" + + def test_unknown(self): + assert finish_reason_to_status("some_future_reason") == "completed" + + +class TestDictsAreConsistent: + """Verify the two dicts agree on the mappings they share.""" + + def test_completed_roundtrips(self): + assert FINISH_REASON_TO_STATUS[STATUS_TO_FINISH_REASON["completed"]] == "completed" + + def test_incomplete_roundtrips(self): + assert FINISH_REASON_TO_STATUS[STATUS_TO_FINISH_REASON["incomplete"]] == "incomplete" + + def test_stop_roundtrips(self): + assert STATUS_TO_FINISH_REASON[FINISH_REASON_TO_STATUS["stop"]] == "stop" + + def test_length_roundtrips(self): + assert STATUS_TO_FINISH_REASON[FINISH_REASON_TO_STATUS["length"]] == "length" + + +class TestResponseFormatToTextFormat: + def test_json_schema(self): + response_format = { + "type": "json_schema", + "json_schema": { + "name": "my_schema", + "schema": {"type": "object"}, + "strict": True, + }, + } + result = response_format_to_text_format(response_format) + assert result == { + "format": { + "type": "json_schema", + "name": "my_schema", + "schema": {"type": "object"}, + "strict": True, + } + } + + def test_json_object(self): + result = response_format_to_text_format({"type": "json_object"}) + assert result == {"format": {"type": "json_object"}} + + def test_text(self): + result = response_format_to_text_format({"type": "text"}) + assert result == {"format": {"type": "text"}} + + def test_none(self): + assert response_format_to_text_format(None) is None + + def test_empty_dict(self): + assert response_format_to_text_format({}) is None + + def test_unknown_type(self): + assert response_format_to_text_format({"type": "unknown"}) is None + + def test_json_schema_defaults(self): + result = response_format_to_text_format({"type": "json_schema"}) + assert result["format"]["name"] == "response_schema" + assert result["format"]["schema"] == {} + assert result["format"]["strict"] is False + + +class TestTextFormatToResponseFormat: + def test_json_schema(self): + text_param = { + "format": { + "type": "json_schema", + "name": "my_schema", + "schema": {"type": "object"}, + "strict": True, + } + } + result = text_format_to_response_format(text_param) + assert result == { + "type": "json_schema", + "json_schema": { + "name": "my_schema", + "schema": {"type": "object"}, + "strict": True, + }, + } + + def test_json_object(self): + result = text_format_to_response_format({"format": {"type": "json_object"}}) + assert result == {"type": "json_object"} + + def test_text_returns_none(self): + """text is the implicit default in CC, so returns None.""" + result = text_format_to_response_format({"format": {"type": "text"}}) + assert result is None + + def test_none(self): + assert text_format_to_response_format(None) is None + + def test_empty_dict(self): + assert text_format_to_response_format({}) is None + + def test_json_schema_defaults(self): + result = text_format_to_response_format( + {"format": {"type": "json_schema"}} + ) + assert result["json_schema"]["name"] == "response_schema" + assert result["json_schema"]["schema"] == {} + assert result["json_schema"]["strict"] is False + + +class TestFormatRoundtrip: + """Verify json_schema and json_object survive a roundtrip.""" + + def test_json_schema_cc_to_responses_to_cc(self): + original = { + "type": "json_schema", + "json_schema": { + "name": "test", + "schema": {"type": "object", "properties": {"x": {"type": "int"}}}, + "strict": True, + }, + } + responses_fmt = response_format_to_text_format(original) + back = text_format_to_response_format(responses_fmt) + assert back == original + + def test_json_object_roundtrip(self): + original = {"type": "json_object"} + responses_fmt = response_format_to_text_format(original) + back = text_format_to_response_format(responses_fmt) + assert back == original + + +class TestNormalizeProviderSpecificFields: + def test_dict_with_field(self): + obj = {"provider_specific_fields": {"key": "value"}} + assert normalize_provider_specific_fields(obj) == {"key": "value"} + + def test_dict_without_field(self): + assert normalize_provider_specific_fields({"other": 1}) is None + + def test_dict_with_none_field(self): + assert normalize_provider_specific_fields({"provider_specific_fields": None}) is None + + def test_dict_with_empty_dict(self): + assert normalize_provider_specific_fields({"provider_specific_fields": {}}) is None + + def test_object_with_attr(self): + class Obj: + provider_specific_fields = {"key": "value"} + assert normalize_provider_specific_fields(Obj()) == {"key": "value"} + + def test_object_without_attr(self): + class Obj: + pass + assert normalize_provider_specific_fields(Obj()) is None + + def test_object_with_none_attr(self): + class Obj: + provider_specific_fields = None + assert normalize_provider_specific_fields(Obj()) is None + + def test_none_input(self): + assert normalize_provider_specific_fields(None) is None + + +class TestResponseApiUsageToChatUsage: + def test_none(self): + result = response_api_usage_to_chat_usage(None) + assert result.prompt_tokens == 0 + assert result.completion_tokens == 0 + assert result.total_tokens == 0 + + def test_basic(self): + usage = ResponseAPIUsage(input_tokens=10, output_tokens=20, total_tokens=30) + result = response_api_usage_to_chat_usage(usage) + assert result.prompt_tokens == 10 + assert result.completion_tokens == 20 + assert result.total_tokens == 30 + + def test_from_dict(self): + result = response_api_usage_to_chat_usage( + {"input_tokens": 5, "output_tokens": 15, "total_tokens": 20} + ) + assert result.prompt_tokens == 5 + assert result.completion_tokens == 15 + + def test_dict_computes_total_tokens(self): + result = response_api_usage_to_chat_usage( + {"input_tokens": 5, "output_tokens": 15} + ) + assert result.total_tokens == 20 + + def test_preserves_cost(self): + usage = ResponseAPIUsage(input_tokens=10, output_tokens=20, total_tokens=30) + usage.cost = 0.05 # type: ignore + result = response_api_usage_to_chat_usage(usage) + assert getattr(result, "cost") == 0.05 + + +class TestChatUsageToResponseApiUsage: + def test_from_usage(self): + usage = Usage(prompt_tokens=10, completion_tokens=20, total_tokens=30) + result = chat_usage_to_response_api_usage(usage) + assert result.input_tokens == 10 + assert result.output_tokens == 20 + assert result.total_tokens == 30 + + def test_from_model_response(self): + resp = ModelResponse() + resp.usage = Usage(prompt_tokens=5, completion_tokens=15, total_tokens=20) + result = chat_usage_to_response_api_usage(resp) + assert result.input_tokens == 5 + assert result.output_tokens == 15 + + def test_none_usage_on_model_response(self): + resp = ModelResponse() + resp.usage = None # type: ignore + result = chat_usage_to_response_api_usage(resp) + assert result.input_tokens == 0 + assert result.output_tokens == 0 + + def test_preserves_cost(self): + usage = Usage(prompt_tokens=10, completion_tokens=20, total_tokens=30) + usage.cost = 0.05 # type: ignore + result = chat_usage_to_response_api_usage(usage) + assert getattr(result, "cost") == 0.05 + + +class TestUsageRoundtrip: + def test_basic_roundtrip_responses_to_cc_to_responses(self): + original = ResponseAPIUsage(input_tokens=10, output_tokens=20, total_tokens=30) + cc = response_api_usage_to_chat_usage(original) + back = chat_usage_to_response_api_usage(cc) + assert back.input_tokens == 10 + assert back.output_tokens == 20 + assert back.total_tokens == 30 + + def test_basic_roundtrip_cc_to_responses_to_cc(self): + original = Usage(prompt_tokens=10, completion_tokens=20, total_tokens=30) + resp = chat_usage_to_response_api_usage(original) + back = response_api_usage_to_chat_usage(resp) + assert back.prompt_tokens == 10 + assert back.completion_tokens == 20 + assert back.total_tokens == 30