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

fix logging model transparency type checks and tests

Browse source
This commit is contained in:
Het1819 2026-06-04 19:02:45 -04:00
parent db4c984cd6
commit 8336571401
3 changed files with 412 additions and 142 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

199
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,46 +461,56 @@ 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) -> dict:
def _calculate_model_transparency(
self, response_obj: Any = None
) -> ModelTransparencyData:
"""
Calculates the requested vs resolved vs response model mismatch metadata.
"""
requested_model: str = getattr(self, "model", "") or ""
resolved_model: str = (
getattr(self, "litellm_params", {}).get("model", requested_model) or ""
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: str = ""
if response_obj:
if hasattr(response_obj, "model"):
response_model = getattr(response_obj, "model", "") or ""
elif isinstance(response_obj, dict):
response_model = response_obj.get("model", "") or ""
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)
# Explicitly type hint as Union[str, bool] to prevent mypy inference errors
model_mismatch: Union[str, bool] = False
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:
if (
resolved_model not in response_model
and response_model not in resolved_model
):
model_mismatch = "resolved_vs_response_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"
has_usage: bool = False
if response_obj:
if hasattr(response_obj, "usage") and getattr(response_obj, "usage"):
has_usage = True
elif isinstance(response_obj, dict) and response_obj.get("usage"):
has_usage = True
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": "upstream" if has_usage else "missing",
"usage_source": usage_source,
}
def process_dynamic_callbacks(self):
@ -523,15 +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,
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 = []
@ -5693,8 +5740,8 @@ def get_standard_logging_object_payload( # noqa: PLR0915
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
)
@ -5834,17 +5881,15 @@ def get_standard_logging_object_payload( # noqa: PLR0915
# emit_standard_logging_payload(payload) - Moved to success_handler to prevent double emitting
if logging_obj is not None:
# Use typing.cast to bypass strict union mismatch errors down the line
safe_response_obj = cast(Any, init_response_obj)
transparency_data = logging_obj._calculate_model_transparency(
response_obj=safe_response_obj
response_obj=init_response_obj
)
payload["requested_model"] = transparency_data.get("requested_model", "")
payload["resolved_model"] = transparency_data.get("resolved_model", "")
payload["response_model"] = transparency_data.get("response_model", "")
payload["model_mismatch"] = transparency_data.get("model_mismatch", False)
payload["usage_source"] = transparency_data.get("usage_source", "missing")
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
@ -5975,87 +6020,7 @@ def _get_traceback_str_for_error(error_str: str) -> str:
# used for unit testing
from typing import Any, Dict, List, Optional, Union
# def create_dummy_standard_logging_payload() -> StandardLoggingPayload:
# # First create the nested objects with proper typing
# model_info = StandardLoggingModelInformation(
# model_map_key="gpt-3.5-turbo", model_map_value=None
# )
# 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_org_id=None,
# spend_logs_metadata=None,
# requester_ip_address=str("127.0.0.1"),
# requester_metadata=None,
# user_api_key_end_user_id=str("test_end_user"),
# )
# hidden_params = StandardLoggingHiddenParams(
# model_id=None,
# cache_key=None,
# api_base=None,
# response_cost=None,
# additional_headers=None,
# litellm_overhead_time_ms=None,
# batch_models=None,
# litellm_model_name=None,
# 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")
# # Create messages and response with proper typing
# messages: List[Dict[str, str]] = [{"role": "user", "content": "Hello, world!"}]
# response: Dict[str, List[Dict[str, Dict[str, str]]]] = {
# "choices": [{"message": {"content": "Hi there!"}}]
# }
# # Main payload initialization
# return StandardLoggingPayload( # type: ignore
# id=str("test_id"),
# call_type=str("completion"),
# stream=bool(False),
# response_cost=response_cost,
# response_cost_failure_debug_info=None,
# status=str("success"),
# total_tokens=int(
# DEFAULT_MOCK_RESPONSE_PROMPT_TOKEN_COUNT
# + DEFAULT_MOCK_RESPONSE_COMPLETION_TOKEN_COUNT
# ),
# prompt_tokens=int(DEFAULT_MOCK_RESPONSE_PROMPT_TOKEN_COUNT),
# completion_tokens=int(DEFAULT_MOCK_RESPONSE_COMPLETION_TOKEN_COUNT),
# startTime=start_time,
# 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"),
# metadata=metadata,
# cache_hit=bool(False),
# cache_key=None,
# saved_cache_cost=saved_cache_cost,
# request_tags=[],
# end_user=None,
# requester_ip_address=str("127.0.0.1"),
# messages=messages,
# response=response,
# error_str=None,
# model_parameters={"stream": True},
# hidden_params=hidden_params,
# )
def create_dummy_standard_logging_payload() -> StandardLoggingPayload:
# First create the nested objects with proper typing
model_info = StandardLoggingModelInformation(

6
litellm/types/utils.py
View file

@ -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,
@ -2965,7 +2965,7 @@ class StandardLoggingPayload(TypedDict):
hidden_params: StandardLoggingHiddenParams
guardrail_information: Optional[List[StandardLoggingGuardrailInformation]]
standard_built_in_tools_params: Optional[StandardBuiltInToolsParams]
# Add these 5 lines for Issue #29680:
# Added for Issue #29680:
requested_model: NotRequired[str]
resolved_model: NotRequired[str]
response_model: NotRequired[str]

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",