scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-03 02:22:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 81

Merge 8336571401 into 2fe9feda71

Browse source
This commit is contained in:
Het1819 2026-06-10 23:12:49 +08:00 • committed by GitHub
commit 4229fd9167
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 484 additions and 75 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

183
litellm/litellm_core_utils/litellm_logging.py
View file

@ -12,6 +12,7 @@ import time
import traceback
from datetime import datetime as dt_object
from functools import lru_cache
from typing_extensions import TypedDict
from typing import (
TYPE_CHECKING,
Any,
@ -291,6 +292,40 @@ def _get_cached_prometheus_logger():
return _PrometheusLogger
ModelTransparencyMismatch = Union[
Literal["requested_vs_resolved_mismatch", "resolved_vs_response_mismatch"],
bool,
]
UsageSource = Literal["upstream", "missing"]
class ModelTransparencyData(TypedDict):
requested_model: str
resolved_model: str
response_model: str
model_mismatch: ModelTransparencyMismatch
usage_source: UsageSource
ResponseCostCalculatorResult = Union[
CostResponseTypes,
ModelResponseStream,
HttpxBinaryResponseContent,
RerankResponse,
Batch,
FineTuningJob,
ResponsesAPIResponse,
ResponseCompletedEvent,
OpenAIFileObject,
LiteLLMRealtimeStreamLoggingObject,
OpenAIModerationResponse,
SearchResponse,
dict,
list,
]
class Logging(LiteLLMLoggingBaseClass):
global supabaseClient, promptLayerLogger, weightsBiasesLogger, logfireLogger, capture_exception, add_breadcrumb, lunaryLogger, logfireLogger, prometheusLogger, slack_app
custom_pricing: bool = False
@ -426,6 +461,58 @@ class Logging(LiteLLMLoggingBaseClass):
self._defer_async_logging: bool = False
self._enqueue_deferred_logging: Optional[Callable[[], None]] = None
def _calculate_model_transparency(
self, response_obj: Any = None
) -> ModelTransparencyData:
"""
Calculates the requested vs resolved vs response model mismatch metadata.
"""
requested_model = self.model or ""
resolved_model_value: Any = self.litellm_params.get("model", requested_model)
resolved_model = (
resolved_model_value
if isinstance(resolved_model_value, str)
else requested_model
)
response_model_value: Any = None
if isinstance(response_obj, dict):
response_model_value = response_obj.get("model")
elif response_obj is not None:
response_model_value = getattr(response_obj, "model", None)
response_model = (
response_model_value if isinstance(response_model_value, str) else ""
)
model_mismatch: ModelTransparencyMismatch = False
if requested_model != resolved_model:
model_mismatch = "requested_vs_resolved_mismatch"
elif (
response_model
and resolved_model
and resolved_model not in response_model
and response_model not in resolved_model
):
model_mismatch = "resolved_vs_response_mismatch"
usage_value: Any = None
if isinstance(response_obj, dict):
usage_value = response_obj.get("usage")
elif response_obj is not None:
usage_value = getattr(response_obj, "usage", None)
usage_source: UsageSource = "upstream" if usage_value is not None else "missing"
return {
"requested_model": requested_model,
"resolved_model": resolved_model,
"response_model": response_model,
"model_mismatch": model_mismatch,
"usage_source": usage_source,
}
def process_dynamic_callbacks(self):
"""
Initializes CustomLogger compatible callbacks in self.dynamic_* callbacks
@ -481,13 +568,17 @@ class Logging(LiteLLMLoggingBaseClass):
isinstance(callback, str)
and callback in litellm._known_custom_logger_compatible_callbacks
):
compatible_callback = cast(
_custom_logger_compatible_callbacks_literal, callback
)
callback_class = _init_custom_logger_compatible_class(
callback, internal_usage_cache=None, llm_router=None # type: ignore
logging_integration=compatible_callback,
internal_usage_cache=None,
llm_router=None, # type: ignore
)
if callback_class is not None:
processed_list.append(callback_class)
# If processing dynamic_success_callbacks, add to dynamic_async_success_callbacks
if dynamic_callbacks_type == "success":
if self.dynamic_async_success_callbacks is None:
self.dynamic_async_success_callbacks = []
@ -1147,12 +1238,12 @@ class Logging(LiteLLMLoggingBaseClass):
if self.litellm_request_debug:
verbose_logger.warning( # .warning ensures this shows up in all environments
"POST Request Sent from LiteLLM",
extra={"api_base": {api_base}, **masked_headers},
extra={"api_base": api_base, **masked_headers},
)
else:
verbose_logger.debug(
"POST Request Sent from LiteLLM",
extra={"api_base": {api_base}, **masked_headers},
extra={"api_base": api_base, **masked_headers},
)
else:
headers = additional_args.get("headers", {})
@ -1850,7 +1941,11 @@ class Logging(LiteLLMLoggingBaseClass):
self.model_call_details["litellm_params"].setdefault("metadata", {})
if self.model_call_details["litellm_params"]["metadata"] is None:
self.model_call_details["litellm_params"]["metadata"] = {}
self.model_call_details["litellm_params"]["metadata"]["hidden_params"] = getattr(logging_result, "_hidden_params", {}) # type: ignore
self.model_call_details["litellm_params"]["metadata"][
"hidden_params"
] = getattr(
logging_result, "_hidden_params", {}
) # type: ignore
if self.model_call_details.get("cache_hit") is True:
self.model_call_details["response_cost"] = 0.0
@ -1929,7 +2024,9 @@ class Logging(LiteLLMLoggingBaseClass):
)
result = result.model_copy()
transformed_usage = TranscriptionUsageObjectTransformation.transform_transcription_usage_object(result.usage) # type: ignore
transformed_usage = TranscriptionUsageObjectTransformation.transform_transcription_usage_object(
result.usage
) # type: ignore
setattr(result, "usage", transformed_usage)
return result
@ -4505,7 +4602,7 @@ def _maybe_auto_initialize_arize_phoenix(_in_memory_loggers: list) -> None:
litellm.logging_callback_manager.add_litellm_callback(phoenix_logger)
verbose_logger.info(
"Auto-initialized Arize Phoenix logger alongside otel " "(endpoint=%s)",
"Auto-initialized Arize Phoenix logger alongside otel (endpoint=%s)",
arize_phoenix_config.endpoint,
)
except Exception as e:
@ -5555,7 +5652,7 @@ def _extract_response_obj_and_hidden_params(
return response_obj, hidden_params
def get_standard_logging_object_payload(
def get_standard_logging_object_payload( # noqa: PLR0915
kwargs: Optional[dict],
init_response_obj: Union[Any, BaseModel, dict],
start_time: dt_object,
@ -5643,7 +5740,8 @@ def get_standard_logging_object_payload(
id = f"{id}_cache_hit{time.time()}" # do not duplicate the request id
saved_cache_cost = (
logging_obj._response_cost_calculator(
result=init_response_obj, cache_hit=False # type: ignore
result=cast(ResponseCostCalculatorResult, init_response_obj),
cache_hit=False,
)
or 0.0
)
@ -5782,7 +5880,19 @@ def get_standard_logging_object_payload(
# emit_standard_logging_payload(payload) - Moved to success_handler to prevent double emitting
if logging_obj is not None:
transparency_data = logging_obj._calculate_model_transparency(
response_obj=init_response_obj
)
payload["requested_model"] = transparency_data["requested_model"]
payload["resolved_model"] = transparency_data["resolved_model"]
payload["response_model"] = transparency_data["response_model"]
payload["model_mismatch"] = transparency_data["model_mismatch"]
payload["usage_source"] = transparency_data["usage_source"]
return payload
except Exception as e:
verbose_logger.exception(
"Error creating standard logging object - {}".format(str(e))
@ -5907,8 +6017,6 @@ def _get_traceback_str_for_error(error_str: str) -> str:
return traceback.format_exc()
from decimal import Decimal
# used for unit testing
from typing import Any, Dict, List, Optional, Union
@ -5920,16 +6028,16 @@ def create_dummy_standard_logging_payload() -> StandardLoggingPayload:
)
metadata = StandardLoggingMetadata( # type: ignore
user_api_key_hash=str("test_hash"),
user_api_key_alias=str("test_alias"),
user_api_key_team_id=str("test_team"),
user_api_key_user_id=str("test_user"),
user_api_key_team_alias=str("test_team_alias"),
user_api_key_hash="test_hash",
user_api_key_alias="test_alias",
user_api_key_team_id="test_team",
user_api_key_user_id="test_user",
user_api_key_team_alias="test_team_alias",
user_api_key_org_id=None,
spend_logs_metadata=None,
requester_ip_address=str("127.0.0.1"),
requester_ip_address="127.0.0.1",
requester_metadata=None,
user_api_key_end_user_id=str("test_end_user"),
user_api_key_end_user_id="test_end_user",
)
hidden_params = StandardLoggingHiddenParams(
@ -5944,12 +6052,12 @@ def create_dummy_standard_logging_payload() -> StandardLoggingPayload:
usage_object=None,
)
# Convert numeric values to appropriate types
response_cost = Decimal("0.1")
start_time = Decimal("1234567890.0")
end_time = Decimal("1234567891.0")
completion_start_time = Decimal("1234567890.5")
saved_cache_cost = Decimal("0.0")
# FIX: Use native floats instead of Decimal
response_cost = 0.1
start_time = 1234567890.0
end_time = 1234567891.0
completion_start_time = 1234567890.5
saved_cache_cost = 0.0
# Create messages and response with proper typing
messages: List[Dict[str, str]] = [{"role": "user", "content": "Hello, world!"}]
@ -5959,12 +6067,12 @@ def create_dummy_standard_logging_payload() -> StandardLoggingPayload:
# Main payload initialization
return StandardLoggingPayload( # type: ignore
id=str("test_id"),
call_type=str("completion"),
stream=bool(False),
id="test_id", # FIX: remove str() wrappers
call_type="completion",
stream=False,
response_cost=response_cost,
response_cost_failure_debug_info=None,
status=str("success"),
status="success", # FIX: Literal expects "success", not str("success")
total_tokens=int(
DEFAULT_MOCK_RESPONSE_PROMPT_TOKEN_COUNT
+ DEFAULT_MOCK_RESPONSE_COMPLETION_TOKEN_COUNT
@ -5975,21 +6083,26 @@ def create_dummy_standard_logging_payload() -> StandardLoggingPayload:
endTime=end_time,
completionStartTime=completion_start_time,
model_map_information=model_info,
model=str("gpt-3.5-turbo"),
model_id=str("model-123"),
model_group=str("openai-gpt"),
custom_llm_provider=str("openai"),
api_base=str("https://api.openai.com"),
model="gpt-3.5-turbo",
model_id="model-123",
model_group="openai-gpt",
custom_llm_provider="openai",
api_base="https://api.openai.com",
metadata=metadata,
cache_hit=bool(False),
cache_hit=False,
cache_key=None,
saved_cache_cost=saved_cache_cost,
request_tags=[],
end_user=None,
requester_ip_address=str("127.0.0.1"),
requester_ip_address="127.0.0.1",
messages=messages,
response=response,
error_str=None,
model_parameters={"stream": True},
hidden_params=hidden_params,
requested_model="gpt-3.5-turbo",
resolved_model="gpt-3.5-turbo",
response_model="gpt-3.5-turbo-0613",
model_mismatch=False,
usage_source="upstream",
)

27
litellm/types/utils.py
View file

@ -39,7 +39,7 @@ from pydantic import (
PrivateAttr,
field_validator,
)
from typing_extensions import Required, TypedDict
from typing_extensions import Required, TypedDict, NotRequired
from litellm._uuid import uuid
from litellm.types.llms.base import (
@ -1161,7 +1161,7 @@ class Message(SafeAttributeModel, OpenAIObject):
provider_specific_fields: Optional[Dict[str, Any]] = Field(default=None)
annotations: Optional[List[ChatCompletionAnnotation]] = None
def __init__(
def __init__( # noqa: PLR0915
self,
content: Optional[str] = None,
role: Literal["assistant", "user", "system", "tool", "function"] = "assistant",
@ -1284,7 +1284,7 @@ class Delta(SafeAttributeModel, OpenAIObject):
reasoning_items: Optional[List[ChatCompletionReasoningItem]] = None
provider_specific_fields: Optional[Dict[str, Any]] = Field(default=None)
def __init__(
def __init__( # noqa: PLR0915
self,
content=None,
role=None,
@ -2727,22 +2727,7 @@ class StandardLoggingPayloadErrorInformation(TypedDict, total=False):
llm_provider: Optional[str]
traceback: Optional[str]
error_message: Optional[str]
# error_rate_limit_category:
# For 429 / rate-limit errors, the source of the rate limit. One of the
# string values defined by `litellm.exceptions.RateLimitErrorCategory`
# (vendor_rate_limit, vendor_batch_rate_limit, litellm_rate_limit,
# litellm_batch_rate_limit). None for non-rate-limit exceptions.
# Surfaced here so custom callbacks / metrics consumers can switch on
# the rate-limit source without reaching for the raw exception.
error_rate_limit_category: Optional[str]
# error_rate_limit_type:
# For 429 / rate-limit errors, the dimension that was exceeded. One of
# the string values defined by `litellm.exceptions.RateLimitType`
# (requests, tokens, concurrent_requests, budget, max_iterations).
# None for non-rate-limit exceptions and for rate-limit exceptions that
# did not classify the failure (e.g. legacy vendor 429 with no header
# hints). Lets dashboards split rate-limit failures by cause without
# parsing free-text error messages.
error_rate_limit_type: Optional[str]
@ -2980,6 +2965,12 @@ class StandardLoggingPayload(TypedDict):
hidden_params: StandardLoggingHiddenParams
guardrail_information: Optional[List[StandardLoggingGuardrailInformation]]
standard_built_in_tools_params: Optional[StandardBuiltInToolsParams]
# Added for Issue #29680:
requested_model: NotRequired[str]
resolved_model: NotRequired[str]
response_model: NotRequired[str]
model_mismatch: NotRequired[Union[str, bool]]
usage_source: NotRequired[str]
from typing import AsyncIterator, Iterator

349
tests/logging_callback_tests/test_standard_logging_payload.py
View file

@ -2,11 +2,9 @@
Unit tests for StandardLoggingPayloadSetup
"""
import json
import os
import sys
from datetime import datetime
from unittest.mock import AsyncMock
sys.path.insert(
0, os.path.abspath("../..")
@ -23,7 +21,6 @@ from litellm.types.utils import (
StandardLoggingHiddenParams,
)
from create_mock_standard_logging_payload import (
create_standard_logging_payload,
create_standard_logging_payload_with_long_content,
)
from litellm.litellm_core_utils.litellm_logging import (
@ -80,6 +77,124 @@ def test_get_usage(response_obj, expected_values):
assert usage.total_tokens == expected_values[2]
def test_calculate_model_transparency_for_object_response():
from types import SimpleNamespace
from litellm.litellm_core_utils.litellm_logging import Logging
logging_obj = Logging(
model="gpt-4o",
messages=[{"role": "user", "content": "Hello"}],
stream=False,
call_type="completion",
start_time=datetime.now(),
litellm_call_id="test-call-id",
function_id="test-function",
)
logging_obj.litellm_params = {"model": "gpt-4o"}
response_obj = SimpleNamespace(
model="claude-3-haiku",
usage={"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
)
result = logging_obj._calculate_model_transparency(response_obj=response_obj)
assert result["requested_model"] == "gpt-4o"
assert result["resolved_model"] == "gpt-4o"
assert result["response_model"] == "claude-3-haiku"
assert result["model_mismatch"] == "resolved_vs_response_mismatch"
assert result["usage_source"] == "upstream"
def test_calculate_model_transparency_for_dict_response_requested_resolved_mismatch():
from litellm.litellm_core_utils.litellm_logging import Logging
logging_obj = Logging(
model="gpt-4",
messages=[{"role": "user", "content": "Hello"}],
stream=False,
call_type="completion",
start_time=datetime.now(),
litellm_call_id="test-call-id",
function_id="test-function",
)
logging_obj.litellm_params = {"model": "azure/gpt-4-deployment"}
response_obj = {
"model": "gpt-4-0613",
"usage": {
"prompt_tokens": 1,
"completion_tokens": 1,
"total_tokens": 2,
},
}
result = logging_obj._calculate_model_transparency(response_obj=response_obj)
assert result["requested_model"] == "gpt-4"
assert result["resolved_model"] == "azure/gpt-4-deployment"
assert result["response_model"] == "gpt-4-0613"
assert result["model_mismatch"] == "requested_vs_resolved_mismatch"
assert result["usage_source"] == "upstream"
def test_standard_logging_payload_includes_model_transparency_fields():
from litellm.litellm_core_utils.litellm_logging import (
Logging,
get_standard_logging_object_payload,
)
logging_obj = Logging(
model="gpt-4",
messages=[{"role": "user", "content": "Hello"}],
stream=False,
call_type="completion",
start_time=datetime.now(),
litellm_call_id="test-call-id",
function_id="test-function",
)
logging_obj.litellm_params = {"model": "azure/gpt-4-deployment"}
response_obj = {
"id": "chatcmpl-test",
"object": "chat.completion",
"model": "gpt-4-0613",
"usage": {
"prompt_tokens": 1,
"completion_tokens": 1,
"total_tokens": 2,
},
"choices": [
{
"index": 0,
"message": {"role": "assistant", "content": "Hello!"},
"finish_reason": "stop",
}
],
}
payload = get_standard_logging_object_payload(
kwargs={
"model": "gpt-4",
"messages": [{"role": "user", "content": "Hello"}],
"response_cost": 0.0,
"custom_llm_provider": "openai",
},
init_response_obj=response_obj,
start_time=datetime.now(),
end_time=datetime.now(),
logging_obj=logging_obj,
status="success",
)
assert payload["requested_model"] == "gpt-4"
assert payload["resolved_model"] == "azure/gpt-4-deployment"
assert payload["response_model"] == "gpt-4-0613"
assert payload["model_mismatch"] == "requested_vs_resolved_mismatch"
assert payload["usage_source"] == "upstream"
def test_get_usage_from_image_generation_response():
"""
Test that image generation usage (with input_tokens/output_tokens format)
@ -177,6 +292,215 @@ def test_get_additional_headers():
)
def test_calculate_model_transparency_missing_usage_no_mismatch():
from litellm.litellm_core_utils.litellm_logging import Logging
logging_obj = Logging(
model="gpt-4o",
messages=[{"role": "user", "content": "Hello"}],
stream=False,
call_type="completion",
start_time=datetime.now(),
litellm_call_id="test-call-id",
function_id="test-function",
)
logging_obj.litellm_params = {"model": "gpt-4o"}
response_obj = {
"model": "gpt-4o",
}
result = logging_obj._calculate_model_transparency(response_obj=response_obj)
assert result["requested_model"] == "gpt-4o"
assert result["resolved_model"] == "gpt-4o"
assert result["response_model"] == "gpt-4o"
assert result["model_mismatch"] is False
assert result["usage_source"] == "missing"
def test_calculate_model_transparency_response_model_missing():
from litellm.litellm_core_utils.litellm_logging import Logging
logging_obj = Logging(
model="gpt-4o",
messages=[{"role": "user", "content": "Hello"}],
stream=False,
call_type="completion",
start_time=datetime.now(),
litellm_call_id="test-call-id",
function_id="test-function",
)
logging_obj.litellm_params = {"model": "gpt-4o"}
response_obj = {
"usage": {
"prompt_tokens": 1,
"completion_tokens": 1,
"total_tokens": 2,
}
}
result = logging_obj._calculate_model_transparency(response_obj=response_obj)
assert result["requested_model"] == "gpt-4o"
assert result["resolved_model"] == "gpt-4o"
assert result["response_model"] == ""
assert result["model_mismatch"] is False
assert result["usage_source"] == "upstream"
def test_calculate_model_transparency_non_string_resolved_model_falls_back_to_requested():
from litellm.litellm_core_utils.litellm_logging import Logging
logging_obj = Logging(
model="gpt-4o",
messages=[{"role": "user", "content": "Hello"}],
stream=False,
call_type="completion",
start_time=datetime.now(),
litellm_call_id="test-call-id",
function_id="test-function",
)
logging_obj.litellm_params = {"model": {"deployment": "gpt-4o"}}
response_obj = {
"model": "gpt-4o",
"usage": {
"prompt_tokens": 1,
"completion_tokens": 1,
"total_tokens": 2,
},
}
result = logging_obj._calculate_model_transparency(response_obj=response_obj)
assert result["requested_model"] == "gpt-4o"
assert result["resolved_model"] == "gpt-4o"
assert result["response_model"] == "gpt-4o"
assert result["model_mismatch"] is False
assert result["usage_source"] == "upstream"
def test_calculate_model_transparency_none_response_object():
from litellm.litellm_core_utils.litellm_logging import Logging
logging_obj = Logging(
model="gpt-4o",
messages=[{"role": "user", "content": "Hello"}],
stream=False,
call_type="completion",
start_time=datetime.now(),
litellm_call_id="test-call-id",
function_id="test-function",
)
logging_obj.litellm_params = {"model": "gpt-4o"}
result = logging_obj._calculate_model_transparency(response_obj=None)
assert result["requested_model"] == "gpt-4o"
assert result["resolved_model"] == "gpt-4o"
assert result["response_model"] == ""
assert result["model_mismatch"] is False
assert result["usage_source"] == "missing"
def test_calculate_model_transparency_empty_requested_model():
from litellm.litellm_core_utils.litellm_logging import Logging
logging_obj = Logging(
model="",
messages=[{"role": "user", "content": "Hello"}],
stream=False,
call_type="completion",
start_time=datetime.now(),
litellm_call_id="test-empty-model",
function_id="test-function",
)
logging_obj.litellm_params = {}
response_obj = {
"model": "",
"usage": {
"prompt_tokens": 1,
"completion_tokens": 1,
"total_tokens": 2,
},
}
result = logging_obj._calculate_model_transparency(response_obj=response_obj)
assert result["requested_model"] == ""
assert result["resolved_model"] == ""
assert result["response_model"] == ""
assert result["model_mismatch"] is False
assert result["usage_source"] == "upstream"
def test_calculate_model_transparency_non_string_response_model():
from litellm.litellm_core_utils.litellm_logging import Logging
logging_obj = Logging(
model="gpt-4o",
messages=[{"role": "user", "content": "Hello"}],
stream=False,
call_type="completion",
start_time=datetime.now(),
litellm_call_id="test-non-string-response-model",
function_id="test-function",
)
logging_obj.litellm_params = {"model": "gpt-4o"}
response_obj = {
"model": {"name": "gpt-4o"},
"usage": {
"prompt_tokens": 1,
"completion_tokens": 1,
"total_tokens": 2,
},
}
result = logging_obj._calculate_model_transparency(response_obj=response_obj)
assert result["requested_model"] == "gpt-4o"
assert result["resolved_model"] == "gpt-4o"
assert result["response_model"] == ""
assert result["model_mismatch"] is False
assert result["usage_source"] == "upstream"
def test_calculate_model_transparency_response_model_with_version_suffix_no_mismatch():
from litellm.litellm_core_utils.litellm_logging import Logging
logging_obj = Logging(
model="gpt-4",
messages=[{"role": "user", "content": "Hello"}],
stream=False,
call_type="completion",
start_time=datetime.now(),
litellm_call_id="test-call-id",
function_id="test-function",
)
logging_obj.litellm_params = {"model": "gpt-4"}
response_obj = {
"model": "gpt-4-0613",
"usage": {
"prompt_tokens": 1,
"completion_tokens": 1,
"total_tokens": 2,
},
}
result = logging_obj._calculate_model_transparency(response_obj=response_obj)
assert result["requested_model"] == "gpt-4"
assert result["resolved_model"] == "gpt-4"
assert result["response_model"] == "gpt-4-0613"
assert result["model_mismatch"] is False
assert result["usage_source"] == "upstream"
def all_fields_present(standard_logging_metadata: StandardLoggingMetadata):
for field in StandardLoggingMetadata.__annotations__.keys():
assert field in standard_logging_metadata
@ -206,8 +530,6 @@ def test_get_standard_logging_metadata(metadata_key, metadata_value):
StandardLoggingPayloadSetup.get_standard_logging_metadata(metadata)
)
print("standard_logging_metadata", standard_logging_metadata)
# Assert that all fields in StandardLoggingMetadata are present
all_fields_present(standard_logging_metadata)
@ -325,7 +647,6 @@ def test_get_model_cost_information():
litellm_info_gpt_3_5_turbo_model_map_value = litellm.get_model_info(
model="gpt-5-mini", custom_llm_provider="openai"
)
print("result", result)
assert result["model_map_key"] == "gpt-5-mini"
assert result["model_map_value"] is not None
assert result["model_map_value"] == litellm_info_gpt_3_5_turbo_model_map_value
@ -392,8 +713,6 @@ def test_get_final_response_obj():
response_obj=model_response, init_response_obj=model_response, kwargs=kwargs
)
print("result", result)
print("type(result)", type(result))
# Verify response message content was redacted
assert result["choices"][0]["message"]["content"] == "redacted-by-litellm"
# Verify that redaction occurred in kwargs
@ -467,11 +786,6 @@ def test_truncate_standard_logging_payload():
assert len_original_response == len(str(original_response))
assert len_original_error_str == len(str(original_error_str))
print(
"logged standard_logging_payload",
json.dumps(standard_logging_payload, indent=2),
)
# Logged messages, response, and error_str should be truncated
# assert len of messages is less than 10_500
assert len(str(standard_logging_payload["messages"])) < 10_500
@ -498,7 +812,6 @@ def test_get_error_information():
# Test with None
result = StandardLoggingPayloadSetup.get_error_information(None)
print("error_information", json.dumps(result, indent=2))
assert result["error_code"] == ""
assert result["error_class"] == ""
assert result["llm_provider"] == ""
@ -506,7 +819,6 @@ def test_get_error_information():
# Test with a basic Exception
basic_exception = Exception("Test error")
result = StandardLoggingPayloadSetup.get_error_information(basic_exception)
print("error_information", json.dumps(result, indent=2))
assert result["error_code"] == ""
assert result["error_class"] == "Exception"
assert result["llm_provider"] == ""
@ -522,7 +834,6 @@ def test_get_error_information():
num_retries=None,
)
result = StandardLoggingPayloadSetup.get_error_information(litellm_exception)
print("error_information", json.dumps(result, indent=2))
assert result["error_code"] == "429"
assert result["error_class"] == "RateLimitError"
assert result["llm_provider"] == "openai"
@ -597,9 +908,7 @@ def test_cost_breakdown_in_standard_logging_payload():
get_standard_logging_object_payload,
Logging,
)
from litellm.types.utils import Usage
from datetime import datetime
import time
# Create a mock logging object with cost breakdown
logging_obj = Logging(
@ -672,8 +981,6 @@ def test_cost_breakdown_in_standard_logging_payload():
assert payload["cost_breakdown"]["total_cost"] == 0.0035
assert payload["response_cost"] == 0.0035
print("✅ Cost breakdown test passed!")
def test_cost_breakdown_missing_in_standard_logging_payload():
"""
@ -731,8 +1038,6 @@ def test_cost_breakdown_missing_in_standard_logging_payload():
assert payload["cost_breakdown"] is None
assert payload["response_cost"] == 0.0001
print("✅ Cost breakdown missing test passed!")
@pytest.mark.parametrize(
"use_combined_usage_object",