tiennm99/litellm
1
0
Fork 0
mirror of https://github.com/tiennm99/litellm.git synced 2026-08-09 12:24:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #24080 from BerriAI/litellm_dev_03_18_2026_p1

Browse Source 
fix: fix logging for response incomplete streaming + custom pricing on /v1/messages and /v1/responses
This commit is contained in:
Krish Dholakia
2026-03-18 21:45:17 -07:00
committed by GitHub
parent 11f0f061bc 08f0cbc2e9
commit 49443cc08c
6 changed files with 513 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
litellm/litellm_core_utils/litellm_logging.py
+26 -2
View File
@@ -84,6 +84,8 @@ from litellm.types.llms.openai import (
OpenAIModerationResponse,
ResponseAPIUsage,
ResponseCompletedEvent,
ResponseFailedEvent,
ResponseIncompleteEvent,
ResponsesAPIResponse,
)
from litellm.types.mcp import MCPPostCallResponseObject
@@ -516,6 +518,23 @@ class Logging(LiteLLMLoggingBaseClass):
),
)
def get_router_model_id(self) -> Optional[str]:
"""Extract the router deployment model_id from litellm_params.
Checks both litellm_metadata and metadata for model_info.id.
Used by cost calculators to look up custom pricing registered
under the deployment's model_info.id in litellm.model_cost.
"""
if not hasattr(self, "litellm_params"):
return None
for key in ("litellm_metadata", "metadata"):
meta = self.litellm_params.get(key, {}) or {}
info = meta.get("model_info", {}) or {}
model_id = info.get("id")
if model_id is not None:
return model_id
return None
def update_environment_variables(
self,
litellm_params: Dict,
@@ -1455,6 +1474,12 @@ class Logging(LiteLLMLoggingBaseClass):
): # use model_id if not already set
router_model_id = hidden_params["model_id"]
# Fallback: extract router_model_id from litellm_params when not available
# from the result object. ResponsesAPIResponse objects (used by /v1/responses
# streaming) don't carry _hidden_params["model_id"] like ModelResponse does.
if router_model_id is None:
router_model_id = self.get_router_model_id()
## RESPONSE COST ##
custom_pricing = use_custom_pricing_for_model(
litellm_params=(
@@ -3307,7 +3332,7 @@ class Logging(LiteLLMLoggingBaseClass):
return result
elif isinstance(result, TextCompletionResponse):
return result
elif isinstance(result, ResponseCompletedEvent):
elif isinstance(result, (ResponseCompletedEvent, ResponseIncompleteEvent, ResponseFailedEvent)):
## return unified Usage object
if isinstance(result.response.usage, ResponseAPIUsage):
transformed_usage = (
@@ -3328,7 +3353,6 @@ class Logging(LiteLLMLoggingBaseClass):
return result.response
else:
return None
return None
def _handle_anthropic_messages_response_logging(self, result: Any) -> ModelResponse:
"""
litellm/proxy/_new_secret_config.yaml
+27 -36
View File
@@ -1,41 +1,32 @@
model_list:
- model_name: gpt-3.5-turbo
litellm_params:
model: openai/gpt-3.5-turbo
api_key: os.environ/OPENAI_API_KEY
- model_name: gpt-4o
litellm_params:
model: openai/gpt-4o
api_key: os.environ/OPENAI_API_KEY
- model_name: claude-sonnet-4-5-20250929
litellm_params:
model: anthropic/claude-sonnet-4-5-20250929
- model_name: gpt-4.1-mini
# OpenAI model for /v1/chat/completions test — 200x custom pricing
- model_name: "gpt-4.1-mini"
litellm_params:
model: openai/gpt-4.1-mini
- model_name: gpt-5-mini
api_key: os.environ/OPENAI_API_KEY
model_info:
id: gpt-4.1-mini-custom-pricing
input_cost_per_token: 0.00004 # 100x standard ($0.40/1M = $0.0000004)
output_cost_per_token: 0.00016 # 100x standard ($1.60/1M = $0.0000016)
# OpenAI model for /v1/responses test — 100x custom pricing
- model_name: "gpt-5"
litellm_params:
model: openai/gpt-5-mini
- model_name: custom_litellm_model
model: openai/gpt-5
api_key: os.environ/OPENAI_API_KEY
model_info:
id: gpt-5-custom-pricing
mode: "chat"
input_cost_per_token: 125 # 100x standard ($1.25/1M = $0.00000125)
output_cost_per_token: 10 # 100x standard ($10.00/1M = $0.00001)
# Anthropic model for /v1/messages test — 100x custom pricing
- model_name: "claude-sonnet-4-20250514"
litellm_params:
model: litellm_agent/claude-sonnet-4-5-20250929
litellm_system_prompt: "Be a helpful assistant."
guardrails:
- guardrail_name: "tool_policy"
litellm_params:
guardrail: tool_policy
mode: [pre_call, post_call]
default_on: true
mcp_servers:
my_http_server:
url: "http://0.0.0.0:8001/mcp"
transport: "http"
description: "My custom MCP server"
available_on_public_internet: true
general_settings:
store_model_in_db: true
store_prompts_in_spend_logs: true
model: anthropic/claude-sonnet-4-20250514
api_key: os.environ/ANTHROPIC_API_KEY
model_info:
id: claude-sonnet-4-custom-pricing
input_cost_per_token: 0.0003 # 100x standard ($0.000003)
output_cost_per_token: 0.0015 # 100x standard ($0.000015)
litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py
+27 -13
View File
@@ -6,20 +6,23 @@ import httpx
import litellm
from litellm._logging import verbose_proxy_logger
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.litellm_logging import \
Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.litellm_logging import \
use_custom_pricing_for_model
from litellm.llms.anthropic import get_anthropic_config
from litellm.llms.anthropic.chat.handler import (
ModelResponseIterator as AnthropicModelResponseIterator,
)
from litellm.llms.anthropic.chat.handler import \
ModelResponseIterator as AnthropicModelResponseIterator
from litellm.proxy._types import PassThroughEndpointLoggingTypedDict
from litellm.proxy.auth.auth_utils import get_end_user_id_from_request_body
from litellm.types.passthrough_endpoints.pass_through_endpoints import (
PassthroughStandardLoggingPayload,
)
from litellm.types.utils import LiteLLMBatch, ModelResponse, TextCompletionResponse
from litellm.types.passthrough_endpoints.pass_through_endpoints import \
PassthroughStandardLoggingPayload
from litellm.types.utils import (LiteLLMBatch, ModelResponse,
TextCompletionResponse)
if TYPE_CHECKING:
from litellm.types.passthrough_endpoints.pass_through_endpoints import EndpointType
from litellm.types.passthrough_endpoints.pass_through_endpoints import \
EndpointType
from ..success_handler import PassThroughEndpointLogging
else:
@@ -124,10 +127,21 @@ class AnthropicPassthroughLoggingHandler:
if custom_llm_provider and not model.startswith(f"{custom_llm_provider}/"):
model_for_cost = f"{custom_llm_provider}/{model}"
router_model_id = logging_obj.get_router_model_id()
custom_pricing = use_custom_pricing_for_model(
litellm_params=(
logging_obj.litellm_params
if hasattr(logging_obj, "litellm_params")
else None
)
)
response_cost = litellm.completion_cost(
completion_response=litellm_model_response,
model=model_for_cost,
custom_llm_provider=custom_llm_provider,
custom_pricing=custom_pricing,
router_model_id=router_model_id,
)
kwargs["response_cost"] = response_cost
@@ -319,9 +333,8 @@ class AnthropicPassthroughLoggingHandler:
import base64
from litellm._uuid import uuid
from litellm.llms.anthropic.batches.transformation import (
AnthropicBatchesConfig,
)
from litellm.llms.anthropic.batches.transformation import \
AnthropicBatchesConfig
from litellm.types.utils import Choices, SpecialEnums
try:
@@ -537,7 +550,8 @@ class AnthropicPassthroughLoggingHandler:
managed_files_hook, "store_unified_object_id"
):
# Create a mock user API key dict for the managed object storage
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy._types import (LitellmUserRoles,
UserAPIKeyAuth)
user_api_key_dict = UserAPIKeyAuth(
user_id=kwargs.get("user_id", "default-user"),
litellm/responses/streaming_iterator.py
+38 -5
View File
@@ -166,11 +166,16 @@ class BaseResponsesAPIStreamingIterator:
)
setattr(item, "encrypted_content", wrapped_content)
# Store the completed response
# Store the completed response (also for incomplete/failed so logging still fires)
_chunk_type = getattr(openai_responses_api_chunk, "type", None)
if (
openai_responses_api_chunk
and getattr(openai_responses_api_chunk, "type", None)
== ResponsesAPIStreamEvents.RESPONSE_COMPLETED
and _chunk_type
in (
ResponsesAPIStreamEvents.RESPONSE_COMPLETED,
ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE,
ResponsesAPIStreamEvents.RESPONSE_FAILED,
)
):
self.completed_response = openai_responses_api_chunk
# Add cost to usage object if include_cost_in_streaming_usage is True
@@ -195,10 +200,12 @@ class BaseResponsesAPIStreamingIterator:
if cost is not None:
setattr(usage_obj, "cost", cost)
except Exception:
# If cost calculation fails, continue without cost
pass
self._handle_logging_completed_response()
if _chunk_type == ResponsesAPIStreamEvents.RESPONSE_FAILED:
self._handle_logging_failed_response()
else:
self._handle_logging_completed_response()
return openai_responses_api_chunk
@@ -216,6 +223,32 @@ class BaseResponsesAPIStreamingIterator:
"""Base implementation - should be overridden by subclasses"""
pass
def _handle_logging_failed_response(self):
"""
Handle logging for RESPONSE_FAILED events by routing to failure handlers.
Unlike _handle_logging_completed_response (which calls success handlers),
this constructs an exception from the response error and routes to
async_failure_handler / failure_handler so logging integrations correctly
record the call as failed.
"""
response_obj = (
getattr(self.completed_response, "response", None)
if self.completed_response
else None
)
error_info = getattr(response_obj, "error", None) if response_obj else None
error_message = "Response failed"
if isinstance(error_info, dict):
error_message = error_info.get("message", str(error_info))
exception = litellm.APIError(
status_code=500,
message=error_message,
llm_provider=self.custom_llm_provider or "",
model=self.model or "",
)
self._handle_failure(exception)
async def _call_post_streaming_deployment_hook(self, chunk):
"""
Allow callbacks to modify streaming chunks before returning (parity with chat).
tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py
+155 -1
View File
@@ -30,9 +30,11 @@ from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterat
from litellm.responses.utils import ResponsesAPIRequestUtils
from litellm.types.llms.openai import (
ResponseCompletedEvent,
ResponseFailedEvent,
ResponseIncompleteEvent,
ResponsesAPIResponse,
ResponsesAPIStreamEvents,
OutputTextDeltaEvent
OutputTextDeltaEvent,
)
@@ -429,3 +431,155 @@ class TestBaseResponsesAPIStreamingIterator:
mock_logging_obj.async_failure_handler.assert_not_called()
mock_logging_obj.failure_handler.assert_not_called()
def test_process_chunk_response_failed_calls_failure_handler(self):
"""
Test that a RESPONSE_FAILED event routes to failure handlers,
not success handlers. Failed responses represent genuine LLM-level
errors and should be logged as failures.
"""
from litellm.responses.streaming_iterator import ResponsesAPIStreamingIterator
mock_response = Mock()
mock_response.headers = {}
mock_response.aiter_lines = Mock()
mock_logging_obj = Mock(spec=LiteLLMLoggingObj)
mock_logging_obj.model_call_details = {"litellm_params": {}}
mock_logging_obj.async_failure_handler = Mock()
mock_logging_obj.failure_handler = Mock()
mock_logging_obj.async_success_handler = Mock()
mock_logging_obj.success_handler = Mock()
mock_config = Mock(spec=BaseResponsesAPIConfig)
mock_responses_api_response = Mock(spec=ResponsesAPIResponse)
mock_responses_api_response.id = "resp_failed_123"
mock_responses_api_response.error = {
"type": "server_error",
"message": "The model encountered an error",
}
mock_responses_api_response.usage = None
mock_failed_event = Mock(spec=ResponseFailedEvent)
mock_failed_event.type = ResponsesAPIStreamEvents.RESPONSE_FAILED
mock_failed_event.response = mock_responses_api_response
mock_config.transform_streaming_response.return_value = mock_failed_event
iterator = ResponsesAPIStreamingIterator(
response=mock_response,
model="gpt-4",
responses_api_provider_config=mock_config,
logging_obj=mock_logging_obj,
litellm_metadata={"model_info": {"id": "model_123"}},
custom_llm_provider="openai",
)
test_chunk_data = {
"type": "response.failed",
"response": {
"id": "resp_failed_123",
"error": {
"type": "server_error",
"message": "The model encountered an error",
},
},
}
with patch.object(
ResponsesAPIRequestUtils,
"_update_responses_api_response_id_with_model_id",
return_value=mock_responses_api_response,
), patch(
"litellm.responses.streaming_iterator.run_async_function"
) as mock_run_async, patch(
"litellm.responses.streaming_iterator.executor"
) as mock_executor:
result = iterator._process_chunk(json.dumps(test_chunk_data))
assert result is not None
assert result.type == ResponsesAPIStreamEvents.RESPONSE_FAILED
assert iterator.completed_response == result
# Failure handler should have been called via _handle_failure
mock_run_async.assert_called_once()
call_kwargs = mock_run_async.call_args
assert (
call_kwargs[1]["async_function"]
== mock_logging_obj.async_failure_handler
)
mock_executor.submit.assert_called_once()
submit_args = mock_executor.submit.call_args
assert submit_args[0][0] == mock_logging_obj.failure_handler
def test_process_chunk_response_incomplete_calls_success_handler(self):
"""
Test that a RESPONSE_INCOMPLETE event routes to success handlers.
Incomplete responses (e.g. max_output_tokens reached) are still valid
responses with usage data — analogous to finish_reason='length' in chat.
"""
from litellm.responses.streaming_iterator import ResponsesAPIStreamingIterator
mock_response = Mock()
mock_response.headers = {}
mock_response.aiter_lines = Mock()
mock_logging_obj = Mock(spec=LiteLLMLoggingObj)
mock_logging_obj.model_call_details = {"litellm_params": {}}
mock_logging_obj.async_failure_handler = Mock()
mock_logging_obj.failure_handler = Mock()
mock_logging_obj.async_success_handler = Mock()
mock_logging_obj.success_handler = Mock()
mock_config = Mock(spec=BaseResponsesAPIConfig)
mock_responses_api_response = Mock(spec=ResponsesAPIResponse)
mock_responses_api_response.id = "resp_incomplete_123"
mock_responses_api_response.incomplete_details = {
"reason": "max_output_tokens"
}
mock_responses_api_response.usage = None
mock_incomplete_event = Mock(spec=ResponseIncompleteEvent)
mock_incomplete_event.type = ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE
mock_incomplete_event.response = mock_responses_api_response
mock_config.transform_streaming_response.return_value = mock_incomplete_event
iterator = ResponsesAPIStreamingIterator(
response=mock_response,
model="gpt-4",
responses_api_provider_config=mock_config,
logging_obj=mock_logging_obj,
litellm_metadata={"model_info": {"id": "model_123"}},
custom_llm_provider="openai",
)
test_chunk_data = {
"type": "response.incomplete",
"response": {
"id": "resp_incomplete_123",
"incomplete_details": {"reason": "max_output_tokens"},
},
}
with patch.object(
ResponsesAPIRequestUtils,
"_update_responses_api_response_id_with_model_id",
return_value=mock_responses_api_response,
), patch(
"asyncio.create_task"
) as mock_create_task, patch(
"litellm.responses.streaming_iterator.executor"
) as mock_executor:
result = iterator._process_chunk(json.dumps(test_chunk_data))
assert result is not None
assert result.type == ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE
assert iterator.completed_response == result
# Success handler should have been called (via _handle_logging_completed_response)
mock_create_task.assert_called_once()
mock_executor.submit.assert_called_once()
# Failure handlers should NOT have been called
mock_logging_obj.async_failure_handler.assert_not_called()
mock_logging_obj.failure_handler.assert_not_called()
tests/test_litellm/litellm_core_utils/test_litellm_logging.py
+240 -26
View File
@@ -11,7 +11,8 @@ sys.path.insert(
import time
from litellm.constants import SENTRY_DENYLIST, SENTRY_PII_DENYLIST
from litellm.litellm_core_utils.litellm_logging import Logging as LitellmLogging
from litellm.litellm_core_utils.litellm_logging import \
Logging as LitellmLogging
from litellm.litellm_core_utils.litellm_logging import set_callbacks
from litellm.types.utils import ModelResponse, TextCompletionResponse
@@ -139,7 +140,8 @@ def test_sentry_environment():
def test_use_custom_pricing_for_model():
from litellm.litellm_core_utils.litellm_logging import use_custom_pricing_for_model
from litellm.litellm_core_utils.litellm_logging import \
use_custom_pricing_for_model
litellm_params = {
"custom_llm_provider": "azure",
@@ -154,7 +156,8 @@ def test_use_custom_pricing_for_model_via_litellm_metadata():
Generic API call routes (/messages, /responses) store model_info
under litellm_metadata, not metadata. Regression test for #23185.
"""
from litellm.litellm_core_utils.litellm_logging import use_custom_pricing_for_model
from litellm.litellm_core_utils.litellm_logging import \
use_custom_pricing_for_model
litellm_params = {
"litellm_metadata": {
@@ -170,7 +173,8 @@ def test_use_custom_pricing_for_model_via_litellm_metadata():
def test_use_custom_pricing_not_detected_litellm_metadata_no_pricing():
"""Should return False when litellm_metadata.model_info has no pricing keys."""
from litellm.litellm_core_utils.litellm_logging import use_custom_pricing_for_model
from litellm.litellm_core_utils.litellm_logging import \
use_custom_pricing_for_model
litellm_params = {
"litellm_metadata": {
@@ -180,6 +184,198 @@ def test_use_custom_pricing_not_detected_litellm_metadata_no_pricing():
assert use_custom_pricing_for_model(litellm_params) is False
def test_response_cost_calculator_uses_router_model_id_from_litellm_metadata():
"""_response_cost_calculator should extract router_model_id from
litellm_params.litellm_metadata.model_info.id when the result object
does not carry _hidden_params (e.g. ResponsesAPIResponse from /v1/responses
streaming). Regression test for custom pricing on streaming responses."""
import litellm
from litellm.litellm_core_utils.litellm_logging import \
Logging as LiteLLMLoggingObj
from litellm.types.llms.openai import ResponsesAPIResponse
custom_model_id = "gpt-5-custom-pricing"
custom_input_cost = 125.0
custom_output_cost = 10.0
litellm.register_model(
model_cost={
custom_model_id: {
"input_cost_per_token": custom_input_cost,
"output_cost_per_token": custom_output_cost,
"max_tokens": 128000,
"max_input_tokens": 128000,
"max_output_tokens": 16384,
"litellm_provider": "openai",
}
}
)
try:
logging_obj = LiteLLMLoggingObj(
model="gpt-5",
messages=[{"role": "user", "content": "Hi"}],
stream=True,
call_type="aresponses",
start_time=time.time(),
litellm_call_id="test-123",
function_id="test-fn",
)
logging_obj.update_environment_variables(
model="gpt-5",
user="",
optional_params={},
litellm_params={
"api_base": "",
"litellm_metadata": {
"model_info": {
"id": custom_model_id,
"input_cost_per_token": custom_input_cost,
"output_cost_per_token": custom_output_cost,
},
},
},
)
response_obj = ResponsesAPIResponse(
id="resp_abc",
created_at=1234567890,
model="gpt-5",
output=[],
usage={
"input_tokens": 10,
"output_tokens": 5,
"total_tokens": 15,
},
)
cost = logging_obj._response_cost_calculator(result=response_obj)
assert cost is not None, "Cost should not be None"
expected_cost = (10 * custom_input_cost) + (5 * custom_output_cost)
assert cost == pytest.approx(
expected_cost
), f"Expected {expected_cost}, got {cost}"
finally:
litellm.model_cost.pop(custom_model_id, None)
class TestGetRouterModelId:
"""Tests for the get_router_model_id helper method."""
def test_returns_id_from_litellm_metadata(self, logging_obj):
"""Should extract model_info.id from litellm_metadata."""
logging_obj.litellm_params = {
"litellm_metadata": {
"model_info": {"id": "custom-deploy-1"},
},
}
assert logging_obj.get_router_model_id() == "custom-deploy-1"
def test_returns_id_from_metadata(self, logging_obj):
"""Should fall back to metadata when litellm_metadata has no model_info."""
logging_obj.litellm_params = {
"metadata": {
"model_info": {"id": "custom-deploy-2"},
},
}
assert logging_obj.get_router_model_id() == "custom-deploy-2"
def test_prefers_litellm_metadata_over_metadata(self, logging_obj):
"""litellm_metadata should take priority over metadata."""
logging_obj.litellm_params = {
"litellm_metadata": {
"model_info": {"id": "from-litellm-meta"},
},
"metadata": {
"model_info": {"id": "from-meta"},
},
}
assert logging_obj.get_router_model_id() == "from-litellm-meta"
def test_returns_none_when_no_model_info(self, logging_obj):
"""Should return None when no model_info is present."""
logging_obj.litellm_params = {"api_base": ""}
assert logging_obj.get_router_model_id() is None
def test_returns_none_when_no_litellm_params(self):
"""Should return None when litellm_params is not set."""
from litellm.litellm_core_utils.litellm_logging import \
Logging as LiteLLMLoggingObj
obj = LiteLLMLoggingObj(
model="test",
messages=[],
stream=False,
call_type="completion",
start_time=time.time(),
litellm_call_id="x",
function_id="x",
)
# litellm_params exists but is empty by default
assert obj.get_router_model_id() is None
class TestAnthropicPassthroughCustomPricing:
"""Verify the Anthropic pass-through handler forwards custom pricing."""
def test_completion_cost_receives_custom_pricing_args(self):
"""_create_anthropic_response_logging_payload should pass
custom_pricing and router_model_id to litellm.completion_cost
when the logging object carries custom pricing in model_info."""
from unittest.mock import patch
from litellm.litellm_core_utils.litellm_logging import \
Logging as LiteLLMLoggingObj
from litellm.proxy.pass_through_endpoints.llm_provider_handlers.anthropic_passthrough_logging_handler import \
AnthropicPassthroughLoggingHandler
logging_obj = LiteLLMLoggingObj(
model="claude-sonnet-4-20250514",
messages=[{"role": "user", "content": "Hi"}],
stream=False,
call_type="anthropic_messages",
start_time=time.time(),
litellm_call_id="test-456",
function_id="test-fn",
)
logging_obj.update_environment_variables(
model="claude-sonnet-4-20250514",
user="",
optional_params={},
litellm_params={
"api_base": "",
"litellm_metadata": {
"model_info": {
"id": "claude-custom-pricing",
"input_cost_per_token": 0.5,
"output_cost_per_token": 1.5,
},
},
},
)
logging_obj.model_call_details["custom_llm_provider"] = "anthropic"
mock_response = ModelResponse()
mock_response.usage = {"prompt_tokens": 10, "completion_tokens": 5} # type: ignore
with patch("litellm.completion_cost", return_value=42.0) as mock_cost:
AnthropicPassthroughLoggingHandler._create_anthropic_response_logging_payload(
litellm_model_response=mock_response,
model="claude-sonnet-4-20250514",
kwargs={},
start_time=time.time(),
end_time=time.time(),
logging_obj=logging_obj,
)
mock_cost.assert_called_once()
call_kwargs = mock_cost.call_args
assert call_kwargs.kwargs.get("custom_pricing") is True
assert call_kwargs.kwargs.get("router_model_id") == "claude-custom-pricing"
class TestUpdateFromKwargs:
"""Tests for the update_from_kwargs convenience wrapper."""
@@ -245,9 +441,8 @@ class TestUpdateFromKwargs:
def test_custom_pricing_detected_via_litellm_metadata(self, logging_obj):
"""Custom pricing in litellm_metadata.model_info should set custom_pricing flag."""
from litellm.litellm_core_utils.litellm_logging import (
use_custom_pricing_for_model,
)
from litellm.litellm_core_utils.litellm_logging import \
use_custom_pricing_for_model
lm_meta = {
"model_info": {
@@ -306,7 +501,8 @@ async def test_datadog_logger_not_shadowed_by_llm_obs(monkeypatch):
monkeypatch.setenv("DD_SITE", "us5.datadoghq.com")
from litellm.integrations.datadog.datadog import DataDogLogger
from litellm.integrations.datadog.datadog_llm_obs import DataDogLLMObsLogger
from litellm.integrations.datadog.datadog_llm_obs import \
DataDogLLMObsLogger
from litellm.litellm_core_utils import litellm_logging as logging_module
logging_module._in_memory_loggers.clear()
@@ -347,7 +543,8 @@ async def test_logfire_logger_accepts_env_vars_for_base_url(monkeypatch):
) # no trailing slash on purpose
# Import after env vars are set (important if module-level caching exists)
from litellm.integrations.opentelemetry import OpenTelemetry # logger class
from litellm.integrations.opentelemetry import \
OpenTelemetry # logger class
from litellm.litellm_core_utils import litellm_logging as logging_module
logging_module._in_memory_loggers.clear()
@@ -676,7 +873,8 @@ def test_success_handler_runs_guardrail_logging_hook_when_enabled(logging_obj):
def test_get_user_agent_tags():
from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup
from litellm.litellm_core_utils.litellm_logging import \
StandardLoggingPayloadSetup
tags = StandardLoggingPayloadSetup._get_user_agent_tags(
proxy_server_request={
@@ -691,7 +889,8 @@ def test_get_user_agent_tags():
def test_get_request_tags():
from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup
from litellm.litellm_core_utils.litellm_logging import \
StandardLoggingPayloadSetup
tags = StandardLoggingPayloadSetup._get_request_tags(
litellm_params={"metadata": {"tags": ["test-tag"]}},
@@ -718,7 +917,8 @@ def test_get_request_tags_from_metadata_and_litellm_metadata():
4. No tags in either
5. None values for metadata/litellm_metadata
"""
from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup
from litellm.litellm_core_utils.litellm_logging import \
StandardLoggingPayloadSetup
# Test case 1: Tags in metadata only
tags = StandardLoggingPayloadSetup._get_request_tags(
@@ -799,7 +999,8 @@ def test_get_request_tags_does_not_mutate_original_tags():
would cause User-Agent tags to be duplicated because the function was mutating
the original tags list instead of creating a copy.
"""
from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup
from litellm.litellm_core_utils.litellm_logging import \
StandardLoggingPayloadSetup
# Create metadata with original tags
original_tags = ["custom-tag-1", "custom-tag-2"]
@@ -859,7 +1060,8 @@ def test_get_request_tags_does_not_mutate_original_tags():
def test_get_extra_header_tags():
"""Test the _get_extra_header_tags method with various scenarios."""
import litellm
from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup
from litellm.litellm_core_utils.litellm_logging import \
StandardLoggingPayloadSetup
# Store original value to restore later
original_extra_headers = getattr(litellm, "extra_spend_tag_headers", None)
@@ -1080,7 +1282,8 @@ async def test_e2e_generate_cold_storage_object_key_successful():
from datetime import datetime, timezone
from unittest.mock import patch
from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup
from litellm.litellm_core_utils.litellm_logging import \
StandardLoggingPayloadSetup
# Create test data
start_time = datetime(2025, 1, 15, 10, 30, 45, 123456, timezone.utc)
@@ -1122,7 +1325,8 @@ async def test_e2e_generate_cold_storage_object_key_with_custom_logger_s3_path()
from datetime import datetime, timezone
from unittest.mock import MagicMock, patch
from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup
from litellm.litellm_core_utils.litellm_logging import \
StandardLoggingPayloadSetup
# Create test data
start_time = datetime(2025, 1, 15, 10, 30, 45, 123456, timezone.utc)
@@ -1173,7 +1377,8 @@ async def test_e2e_generate_cold_storage_object_key_with_logger_no_s3_path():
from datetime import datetime, timezone
from unittest.mock import MagicMock, patch
from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup
from litellm.litellm_core_utils.litellm_logging import \
StandardLoggingPayloadSetup
# Create test data
start_time = datetime(2025, 1, 15, 10, 30, 45, 123456, timezone.utc)
@@ -1220,7 +1425,8 @@ async def test_e2e_generate_cold_storage_object_key_not_configured():
from unittest.mock import patch
import litellm
from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup
from litellm.litellm_core_utils.litellm_logging import \
StandardLoggingPayloadSetup
# Create test data
start_time = datetime(2025, 1, 15, 10, 30, 45, 123456, timezone.utc)
@@ -1244,7 +1450,8 @@ def test_get_final_response_obj_with_empty_response_obj_and_list_init():
When response_obj is empty (falsy), the method should return init_response_obj if it's a list.
"""
from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup
from litellm.litellm_core_utils.litellm_logging import \
StandardLoggingPayloadSetup
# Create test objects
class TestObject1:
@@ -1280,7 +1487,8 @@ def test_get_usage_as_dict():
"""
Test get_usage_as_dict returns usage as plain dict from response_obj or combined_usage_object.
"""
from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup
from litellm.litellm_core_utils.litellm_logging import \
StandardLoggingPayloadSetup
from litellm.types.utils import Usage
# Test case 1: None response_obj returns empty usage dict
@@ -1318,7 +1526,8 @@ def test_append_system_prompt_messages():
"""
Test append_system_prompt_messages prepends system message from kwargs to messages list.
"""
from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup
from litellm.litellm_core_utils.litellm_logging import \
StandardLoggingPayloadSetup
# Test case 1: system in kwargs with existing messages
kwargs = {"system": "You are a helpful assistant"}
@@ -1389,7 +1598,8 @@ async def test_async_success_handler_sets_standard_logging_object_for_pass_throu
from datetime import datetime
from unittest.mock import patch
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.litellm_logging import \
Logging as LiteLLMLoggingObj
from litellm.types.utils import StandardPassThroughResponseObject
# Create a logging object for a pass-through endpoint
@@ -1470,7 +1680,8 @@ async def test_async_success_handler_prevents_reprocessing_for_pass_through_endp
from datetime import datetime
from unittest.mock import patch
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.litellm_logging import \
Logging as LiteLLMLoggingObj
from litellm.types.utils import StandardPassThroughResponseObject
# Create a logging object for a pass-through endpoint
@@ -1546,7 +1757,8 @@ async def test_async_success_handler_sets_standard_logging_object_for_streaming_
from datetime import datetime
from unittest.mock import patch
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.litellm_logging import \
Logging as LiteLLMLoggingObj
from litellm.types.utils import StandardPassThroughResponseObject
# Create a logging object for a streaming pass-through endpoint
@@ -1602,7 +1814,8 @@ def test_get_error_information_error_code_priority():
Test get_error_information prioritizes 'code' attribute over 'status_code' attribute
and handles edge cases like empty strings and "None" string values.
"""
from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup
from litellm.litellm_core_utils.litellm_logging import \
StandardLoggingPayloadSetup
# Test case 1: Exception with 'code' attribute (ProxyException style)
class ProxyException(Exception):
@@ -1795,7 +2008,8 @@ async def test_async_success_handler_preserves_response_cost_for_pass_through_en
by pass-through handlers (Gemini/Vertex)."""
from datetime import datetime
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.litellm_logging import \
Logging as LiteLLMLoggingObj
from litellm.types.utils import ModelResponse, Usage
logging_obj = LiteLLMLoggingObj(