scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-08-28 05:25:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

Revert "Merge pull request #24417 from Chesars/refactor/shared-format-mapping"

Browse source 
This reverts commit 3f3d275e67, reversing
changes made to 498c113933.
This commit is contained in:
Chesars 2026-04-25 15:03:24 -03:00
parent 3c8677209b
commit a4a8d86c2b
5 changed files with 367 additions and 672 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

100
litellm/completion_extras/litellm_responses_transformation/transformation.py
View file

@ -27,10 +27,6 @@ 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,
)
@ -110,7 +106,16 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
# Handle function_call items (e.g., from GPT-5 Codex format)
if item_type == "function_call":
provider_specific_fields = normalize_provider_specific_fields(item)
# 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 {}
)
tool_call_dict = {
"id": item.get("call_id") or item.get("id", ""),
@ -870,8 +875,51 @@ 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 to Responses API text.format."""
return response_format_to_text_format(response_format)
"""
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
@staticmethod
def _convert_annotations_to_chat_format(
@ -913,6 +961,21 @@ 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
@ -992,7 +1055,16 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
# New output item added
output_item = parsed_chunk.get("item", {})
if output_item.get("type") == "function_call":
provider_specific_fields = normalize_provider_specific_fields(output_item)
# 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 {}
)
function_chunk = ChatCompletionToolCallFunctionChunk(
name=output_item.get("name", None),
@ -1057,13 +1129,23 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
# New output item added
output_item = parsed_chunk.get("item", {})
if output_item.get("type") == "function_call":
provider_specific_fields = normalize_provider_specific_fields(output_item)
# 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 {}
)
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"

342
litellm/responses/format_mapping.py
View file

@ -1,342 +0,0 @@
"""
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 None
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):
usage_input = dict(usage_input)
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:
usage_input["total_tokens"] = input_tokens + output_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

217
litellm/responses/litellm_completion_transformation/transformation.py
View file

@ -15,12 +15,6 @@ 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 (
chat_usage_to_response_api_usage,
finish_reason_to_status,
normalize_provider_specific_fields,
text_format_to_response_format,
)
from litellm.types.llms.openai import (
AllMessageValues,
ChatCompletionImageObject,
@ -1468,7 +1462,29 @@ class LiteLLMCompletionResponsesConfig:
for tool in all_chat_completion_tools:
if tool.type == "function":
function_definition = tool.function
provider_specific_fields = normalize_provider_specific_fields(tool) or normalize_provider_specific_fields(function_definition)
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 {}
)
output_tool_call: ResponseFunctionToolCall = ResponseFunctionToolCall(
name=function_definition.name or "",
@ -1490,8 +1506,29 @@ 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."""
return finish_reason_to_status(finish_reason)
"""
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"
@staticmethod
def convert_response_function_tool_call_to_chat_completion_tool_call(
@ -1508,7 +1545,28 @@ class LiteLLMCompletionResponsesConfig:
Returns:
Dictionary in ChatCompletionToolCallChunk format
"""
provider_specific_fields = normalize_provider_specific_fields(tool_call_item)
# 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 {}
)
)
function_dict: Dict[str, Any] = {
"name": tool_call_item.name,
@ -1990,12 +2048,143 @@ class LiteLLMCompletionResponsesConfig:
def _transform_chat_completion_usage_to_responses_usage(
chat_completion_response: Union[ModelResponse, Usage],
) -> ResponseAPIUsage:
"""Transform Chat Completions Usage to ResponseAPIUsage."""
return chat_usage_to_response_api_usage(chat_completion_response)
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,
)
# 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
@staticmethod
def _transform_text_format_to_response_format(
text_param: Union[Dict[str, Any], Any],
) -> Optional[Dict[str, Any]]:
"""Transform Responses API text.format to Chat Completion response_format."""
return text_format_to_response_format(text_param)
"""
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

76
litellm/responses/utils.py
View file

@ -23,7 +23,6 @@ from litellm.types.llms.openai import (
ResponsesAPIResponse,
ResponseText,
)
from litellm.responses.format_mapping import response_api_usage_to_chat_usage
from litellm.types.responses.main import DecodedResponseId
from litellm.types.utils import (
CompletionTokensDetailsWrapper,
@ -652,5 +651,76 @@ class ResponseAPILoggingUtils:
def _transform_response_api_usage_to_chat_usage(
usage_input: Optional[Union[dict, ResponseAPIUsage]],
) -> Usage:
"""Transform ResponseAPIUsage to Chat Completions Usage."""
return response_api_usage_to_chat_usage(usage_input)
"""
Transforms ResponseAPIUsage or ImageUsage to a Usage object.
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

304
tests/test_litellm/responses/test_format_mapping.py
View file

@ -1,304 +0,0 @@
"""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