Skip to content
Merged
5 changes: 4 additions & 1 deletion .github/workflows/test-code-quality.yml
Original file line number Diff line number Diff line change
Expand Up @@ -178,7 +178,7 @@ jobs:
version: "0.10.9"

- name: Install dependencies
run: uv sync --frozen --extra proxy --python 3.10
run: uv sync --frozen --extra proxy --extra cli --python 3.10

- run: uv run --no-sync python --version

Expand All @@ -187,3 +187,6 @@ jobs:

- name: Check litellm CLI
run: uv run --no-sync litellm --version

- name: Check lite CLI
run: uv run --no-sync lite version
35 changes: 25 additions & 10 deletions litellm/caching/caching_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
_assemble_complete_response_from_streaming_chunks,
)
from litellm.types.caching import CachedEmbedding
from litellm.types.integrations.custom_logger import converted_stream_requested
from litellm.types.llms.openai import ResponsesAPIResponse
from litellm.types.rerank import RerankResponse
from litellm.types.utils import (
Expand Down Expand Up @@ -107,17 +108,31 @@ def _is_chat_completion_cached_dict(cached_result: dict) -> bool:
return "choices" in cached_result


def _should_defer_streaming_cache_hit_callbacks(*, kwargs: dict[str, object]) -> bool:
def _stream_replay_requested(kwargs: Mapping[str, object]) -> bool:
if kwargs.get("stream", False) is True:
return True
return converted_stream_requested(kwargs) and not kwargs.get("_agentic_loop_depth")


def _should_defer_streaming_cache_hit_callbacks(*, cached_result: object) -> bool:
"""
When stream=True, do not run success callbacks at cache-hit time.
When the cache hit is replayed as a stream, do not run success callbacks at cache-hit time.

Cached chat/text completion replay uses CustomStreamWrapper; cached Responses
replay uses CachedResponsesAPIStreamingIterator; cached Anthropic Messages
replay uses CachedAnthropicMessagesStreamIterator. All invoke logging success
handlers when the stream finishes; firing them here too would double-count
spend and callback records.
spend and callback records. A plain (non-stream) replay logs here, since nothing
else will.
"""
return kwargs.get("stream", False) is True
from litellm.llms.anthropic.experimental_pass_through.messages.response_cache import (
CachedAnthropicMessagesStreamIterator,
)
from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator

return isinstance(
cached_result, (CustomStreamWrapper, BaseResponsesAPIStreamingIterator, CachedAnthropicMessagesStreamIterator)
)


def _prompt_tokens_details_as_mapping(details: "PromptTokensDetailsWrapper") -> Mapping[str, object]:
Expand Down Expand Up @@ -267,7 +282,7 @@ async def _async_get_cache(
custom_llm_provider=kwargs.get("custom_llm_provider", None),
args=args,
)
if not _should_defer_streaming_cache_hit_callbacks(kwargs=kwargs):
if not _should_defer_streaming_cache_hit_callbacks(cached_result=cached_result):
# LOG SUCCESS
self._async_log_cache_hit_on_callbacks(
logging_obj=logging_obj,
Expand Down Expand Up @@ -383,7 +398,7 @@ def _sync_get_cache(
is_async=False,
)

if not _should_defer_streaming_cache_hit_callbacks(kwargs=kwargs):
if not _should_defer_streaming_cache_hit_callbacks(cached_result=cached_result):
logging_obj.handle_sync_success_callbacks_for_async_calls(
result=cached_result,
start_time=start_time,
Expand Down Expand Up @@ -823,7 +838,7 @@ def _convert_cached_result_to_model_response(
if (call_type == CallTypes.acompletion.value or call_type == CallTypes.completion.value) and isinstance(
cached_result, dict
):
if kwargs.get("stream", False) is True:
if _stream_replay_requested(kwargs):
cached_result = self._convert_cached_stream_response(
cached_result=cached_result,
call_type=call_type,
Expand All @@ -838,7 +853,7 @@ def _convert_cached_result_to_model_response(
if (
call_type == CallTypes.atext_completion.value or call_type == CallTypes.text_completion.value
) and isinstance(cached_result, dict):
if kwargs.get("stream", False) is True:
if _stream_replay_requested(kwargs):
cached_result = self._convert_cached_stream_response(
cached_result=cached_result,
call_type=call_type,
Expand Down Expand Up @@ -893,7 +908,7 @@ def _convert_cached_result_to_model_response(
elif (call_type == "aresponses" or call_type == "responses") and isinstance(cached_result, dict):
use_chat_completion_cache: Final = _is_chat_completion_cached_dict(cached_result)
if use_chat_completion_cache:
if kwargs.get("stream", False) is True:
if _stream_replay_requested(kwargs):
bridge_call_type: Final = (
CallTypes.acompletion.value if call_type == "aresponses" else CallTypes.completion.value
)
Expand Down Expand Up @@ -921,7 +936,7 @@ def _convert_cached_result_to_model_response(
):
response_obj._hidden_params["cache_hit"] = True

if kwargs.get("stream", False) is True:
if _stream_replay_requested(kwargs):
cached_result = CachedResponsesAPIStreamingIterator(
response=response_obj,
logging_obj=logging_obj,
Expand Down
18 changes: 18 additions & 0 deletions litellm/litellm_core_utils/litellm_logging.py
Original file line number Diff line number Diff line change
Expand Up @@ -633,6 +633,24 @@ def set_response_timing_metrics(self, timing_metrics: Mapping[str, float]) -> No
"""Keep ``_response_ms`` / ``litellm_overhead_time_ms`` for a result that has no ``_hidden_params``."""
self.response_timing_metrics = dict(timing_metrics) # mutable-ok: kept deep-copyable

def add_dynamic_callback(self, callback: CustomLogger) -> None:
self.dynamic_input_callbacks = self._with_dynamic_callback(self.dynamic_input_callbacks, callback)
self.dynamic_success_callbacks = self._with_dynamic_callback(self.dynamic_success_callbacks, callback)
self.dynamic_async_success_callbacks = self._with_dynamic_callback(
self.dynamic_async_success_callbacks, callback
)
self.dynamic_failure_callbacks = self._with_dynamic_callback(self.dynamic_failure_callbacks, callback)
self.dynamic_async_failure_callbacks = self._with_dynamic_callback(
self.dynamic_async_failure_callbacks, callback
)

@staticmethod
def _with_dynamic_callback(
callbacks: Sequence[str | Callable | CustomLogger] | None, callback: CustomLogger
) -> list[str | Callable | CustomLogger]:
existing: Final = tuple(callbacks or ())
return [*existing, *(() if callback in existing else (callback,))]

def process_dynamic_callbacks(self):
"""
Initializes CustomLogger compatible callbacks in self.dynamic_* callbacks
Expand Down
8 changes: 1 addition & 7 deletions litellm/proxy/auth/auth_checks.py
Original file line number Diff line number Diff line change
Expand Up @@ -121,6 +121,7 @@
from litellm.repositories.team_repository import TeamRepository
from litellm.repositories.user_repository import UserRepository
from litellm.router import Router
from litellm.types.proxy.auth.auth_checks import UserNotFoundError
from litellm.types.proxy.model_access_group_budget import ModelAccessGroupBudget
from litellm.utils import get_utc_datetime

Expand Down Expand Up @@ -2368,13 +2369,6 @@ async def _backfill_null_user_email(
return updated_row


class UserNotFoundError(ValueError):
"""The user row is provably absent, as opposed to merely unreadable, so a caller that reads a missing row as no user-level limits can key on it without also swallowing a database that would not answer."""

def __init__(self, user_id: str) -> None:
super().__init__(f"User doesn't exist in db. 'user_id'={user_id}. Create user via `/user/new` call.")


@log_db_metrics
async def get_user_object(
user_id: str | None,
Expand Down
4 changes: 2 additions & 2 deletions litellm/proxy/client/cli/commands/pi.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
import tempfile
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from enum import StrEnum
from enum import Enum
from pathlib import Path
from types import MappingProxyType
from typing import Annotated, Final
Expand All @@ -25,7 +25,7 @@
_REJECTED_STATUSES: Final = frozenset((401, 403))


class ListingFailure(StrEnum):
class ListingFailure(str, Enum):
"""Why a proxy could not be listed, decided once where the HTTP outcome is classified.

`unreachable` means no response at all; the other kinds prove the proxy answered, so callers
Expand Down
62 changes: 35 additions & 27 deletions litellm/proxy/common_request_processing.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,11 @@
UNSAFE_PROXY_RESPONSE_HEADERS,
)
from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.litellm_core_utils.core_helpers import get_or_create_metadata_bucket, is_expected_client_error
from litellm.litellm_core_utils.core_helpers import (
get_or_create_metadata_bucket,
independent_snapshot,
is_expected_client_error,
)
from litellm.litellm_core_utils.dd_tracing import NullTracer, tracer
from litellm.litellm_core_utils.get_supported_openai_params import (
get_supported_openai_params,
Expand Down Expand Up @@ -2034,6 +2038,13 @@ async def _pre_call_with_fallbacks(
) -> tuple[dict, LiteLLMLoggingObj]:
from litellm.proxy.common_utils.proxy_rate_limit_error import ProxyRateLimitError

configured_fallbacks: Final = (
self._configured_fallbacks(llm_router=llm_router, user_api_key_dict=user_api_key_dict)
if llm_router is not None and not self.data.get("disable_fallbacks")
else None
)
pristine: Final = independent_snapshot(self.data) if configured_fallbacks else None

try:
return await self.common_processing_pre_call_logic(
request=request,
Expand All @@ -2052,14 +2063,19 @@ async def _pre_call_with_fallbacks(
llm_router=llm_router,
)
except ProxyRateLimitError as original_exc:
original_model: Final = self.data.get("model")
if not original_model or not llm_router or self.data.get("disable_fallbacks"):
rate_limited_data: Final = self.data
original_model: Final = rate_limited_data.get("model")
if (
pristine is None
or not configured_fallbacks
or rate_limited_data.get("disable_fallbacks")
or not isinstance(original_model, str)
):
raise

fallback_models: Final = self._resolve_fallback_models(
model=original_model,
llm_router=llm_router,
user_api_key_dict=user_api_key_dict,
fallbacks=configured_fallbacks,
)
if not fallback_models:
raise
Expand All @@ -2074,6 +2090,7 @@ async def _pre_call_with_fallbacks(
for fallback_model in fallback_models:
if fallback_model == original_model:
continue
self.data = independent_snapshot(pristine)
self.data["model"] = fallback_model
try:
return await self.common_processing_pre_call_logic(
Expand All @@ -2095,39 +2112,30 @@ async def _pre_call_with_fallbacks(
except ProxyRateLimitError:
continue
except BaseException:
self.data["model"] = original_model
self.data = rate_limited_data
raise

self.data["model"] = original_model
self.data = rate_limited_data
raise original_exc

def _resolve_fallback_models(
self,
model: str,
llm_router: Router,
user_api_key_dict: UserAPIKeyAuth,
) -> list | None:
from litellm.router_utils.fallback_event_handlers import get_fallback_model_group

fallbacks = None

@staticmethod
def _configured_fallbacks(llm_router: Router, user_api_key_dict: UserAPIKeyAuth) -> list | None:
key_router_settings: Final = user_api_key_dict.router_settings
if isinstance(key_router_settings, dict) and "fallbacks" in key_router_settings:
fallbacks = key_router_settings["fallbacks"]
key_fallbacks: Final = key_router_settings.get("fallbacks") if isinstance(key_router_settings, dict) else None
fallbacks: Final = key_fallbacks if key_fallbacks is not None else llm_router.fallbacks
return fallbacks if isinstance(fallbacks, list) and fallbacks else None

if fallbacks is None:
fallbacks = llm_router.fallbacks

if not fallbacks:
return None
@staticmethod
def _resolve_fallback_models(model: str, fallbacks: list) -> list | None:
from litellm.router_utils.fallback_event_handlers import get_fallback_model_group

fallback_model_group, generic_fallback_idx = get_fallback_model_group(
fallbacks=fallbacks,
model_group=model,
)
if fallback_model_group is None and generic_fallback_idx is not None:
fallback_model_group = fallbacks[generic_fallback_idx]["*"]
return fallback_model_group
if fallback_model_group is not None:
return fallback_model_group
return fallbacks[generic_fallback_idx]["*"] if generic_fallback_idx is not None else None

@staticmethod
def _get_model_id_from_response(hidden_params: Mapping[str, object], data: Mapping[str, object]) -> str:
Expand Down
5 changes: 5 additions & 0 deletions litellm/proxy/db/routing_prisma_wrapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,11 @@ def __init__(self, db: "PrismaWrapper | RoutingPrismaWrapper") -> None:
self.db: Final = db.writer if isinstance(db, RoutingPrismaWrapper) and not db.writer_unavailable else db


def writer_wrapper(db: "PrismaWrapper | RoutingPrismaWrapper") -> PrismaWrapper:
"""Unlike `WriterPinnedClient`, ignores `writer_unavailable`: a raw SQL write has no replica fallback."""
return db.writer if isinstance(db, RoutingPrismaWrapper) else db


class RoutingPrismaWrapper:
"""
Routes Prisma operations between a writer and a reader Prisma client.
Expand Down
7 changes: 3 additions & 4 deletions litellm/proxy/hooks/responses_id_security.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@
BaseLiteLLMOpenAIResponseObject,
ResponsesAPIResponse,
)
from litellm.types.utils import CallTypesLiteral, LLMResponseTypes, SpecialEnums
from litellm.types.utils import ADDRESSED_RESPONSE_ID_FIELD, CallTypesLiteral, LLMResponseTypes, SpecialEnums

if TYPE_CHECKING:
from litellm.caching.caching import DualCache
Expand All @@ -32,7 +32,6 @@
_RESPONSES_API_PROVIDER_PREFIX: Final = "/openai"
_RESPONSES_API_CREATE_ROUTES: Final = frozenset({"/v1/responses", "/responses"})

_ADDRESSED_RESPONSE_ID_KEY: Final = "_litellm_addressed_response_id"
_UNMANAGED_RESPONSE_ID_DETAIL: Final = (
"Forbidden. This response id was not issued by this proxy, so the proxy cannot tell who owns it. "
"To let keys address responses this proxy did not issue, set "
Expand Down Expand Up @@ -132,15 +131,15 @@ async def async_pre_call_hook(
if call_type not in responses_api_call_types:
return None
addressed_id_field: Final = "previous_response_id" if call_type == "aresponses" else "response_id"
retained_id: Final = data.get(_ADDRESSED_RESPONSE_ID_KEY)
retained_id: Final = data.get(ADDRESSED_RESPONSE_ID_FIELD)
addressed_id: Final = (
retained_id if isinstance(retained_id, str) and retained_id else data.get(addressed_id_field)
)
if not isinstance(addressed_id, str) or not addressed_id:
return data
authorized_id: Final = self._authorize_response_id(addressed_id, user_api_key_dict)
data[addressed_id_field] = authorized_id
data[_ADDRESSED_RESPONSE_ID_KEY] = addressed_id
data[ADDRESSED_RESPONSE_ID_FIELD] = addressed_id
return data

def _authorize_response_id(
Expand Down
Loading
Loading