From 5c76148f717c8037c614b605deef6a679c168f4f Mon Sep 17 00:00:00 2001 From: Tin Date: Sat, 3 Oct 2026 13:01:31 -0700 Subject: [PATCH] fix(shadow-eval): isolate creator billing and budget settlement --- litellm/caching/caching_handler.py | 5 +- litellm/caching/in_memory_cache.py | 2 +- litellm/integrations/lago.py | 12 +- litellm/integrations/shadow_eval_logger.py | 76 +++- litellm/litellm_core_utils/fallback_utils.py | 2 +- .../internal_call_metadata.py | 110 ++++- litellm/litellm_core_utils/litellm_logging.py | 73 +++- litellm/litellm_core_utils/llm_judge.py | 6 +- .../pass_through/messages/handler.py | 52 ++- litellm/main.py | 14 +- litellm/proxy/auth/user_api_key_auth.py | 10 +- .../proxy/hooks/model_max_budget_limiter.py | 16 +- .../proxy/hooks/proxy_track_cost_callback.py | 40 +- .../auto_router_endpoints.py | 6 +- litellm/proxy/native_compaction.py | 6 + litellm/proxy/proxy_server.py | 12 +- .../spend_tracking/budget_reservation.py | 37 +- .../proxy/spend_tracking/evaluation_budget.py | 383 ++++++++++++++++++ litellm/router_strategy/budget_limiter.py | 3 +- litellm/utils.py | 124 +++++- tests/unit/conftest.py | 18 + tests/unit/integrations/test_lago.py | 43 ++ .../integrations/test_shadow_eval_logger.py | 37 +- .../test_litellm_logging.py | 225 ++++++++++ .../unit/litellm_core_utils/test_llm_judge.py | 29 +- .../spend_tracking/test_evaluation_budget.py | 242 +++++++++++ tests/unit/test_utils.py | 344 +++++++++++++++- .../_components/ShadowEvalStartForm.tsx | 4 +- ui/litellm-dashboard/src/lib/http/schema.d.ts | 2 +- 29 files changed, 1827 insertions(+), 106 deletions(-) create mode 100644 litellm/proxy/spend_tracking/evaluation_budget.py create mode 100644 tests/unit/integrations/test_lago.py create mode 100644 tests/unit/proxy/spend_tracking/test_evaluation_budget.py diff --git a/litellm/caching/caching_handler.py b/litellm/caching/caching_handler.py index f04ff8b6e78..f72adf8dbdc 100644 --- a/litellm/caching/caching_handler.py +++ b/litellm/caching/caching_handler.py @@ -738,10 +738,13 @@ class LLMCachingHandler: end_time (datetime): The end time of the operation. cache_hit (bool): Whether it was a cache hit. """ + from litellm.litellm_core_utils.litellm_logging import evaluation_logging_snapshot from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER + receipt_logger: Final = evaluation_logging_snapshot(logging_obj) + GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue( - async_coroutine=logging_obj.async_success_handler( + async_coroutine=receipt_logger.async_success_handler( result=cached_result, start_time=start_time, end_time=end_time, diff --git a/litellm/caching/in_memory_cache.py b/litellm/caching/in_memory_cache.py index 56c9147e066..3728194e070 100644 --- a/litellm/caching/in_memory_cache.py +++ b/litellm/caching/in_memory_cache.py @@ -168,7 +168,7 @@ class InMemoryCache(BaseCache): return self.cache_dict[key] = value - if self.allow_ttl_override(key): # if ttl is not set, set it to default ttl + if kwargs.get("refresh_ttl", False) or self.allow_ttl_override(key): if "ttl" in kwargs and kwargs["ttl"] is not None: self.ttl_dict[key] = self._clock() + float(kwargs["ttl"]) heapq.heappush(self.expiration_heap, (self.ttl_dict[key], key)) diff --git a/litellm/integrations/lago.py b/litellm/integrations/lago.py index 594427b1e0a..b1b78d4f8ba 100644 --- a/litellm/integrations/lago.py +++ b/litellm/integrations/lago.py @@ -3,14 +3,19 @@ import json import os +from collections.abc import Mapping from typing import Final, Literal import httpx +from pydantic import TypeAdapter import litellm from litellm._logging import verbose_logger from litellm._uuid import uuid from litellm.integrations.custom_logger import CustomLogger +from litellm.litellm_core_utils.internal_call_metadata import ( + get_evaluation_billing_owner_from_kwargs, +) from litellm.llms.custom_httpx.http_handler import ( HTTPHandler, get_async_httpx_client, @@ -61,6 +66,9 @@ class LagoLogger(CustomLogger): raise Exception(f"Missing keys={missing_keys} in environment.") def _common_logic(self, kwargs: dict, response_obj) -> dict: + billing_owner: Final = get_evaluation_billing_owner_from_kwargs( + TypeAdapter(Mapping[str, object]).validate_python(kwargs) + ) response_obj.get("id", kwargs.get("litellm_call_id")) get_utc_datetime().isoformat() cost: Final = kwargs.get("response_cost", None) @@ -96,7 +104,9 @@ class LagoLogger(CustomLogger): else: raise Exception("invalid LAGO_API_CHARGE_BY set") - if charge_by == "end_user_id": + if billing_owner is not None: + external_customer_id = billing_owner.user_id + elif charge_by == "end_user_id": external_customer_id = end_user_id elif charge_by == "team_id": external_customer_id = team_id diff --git a/litellm/integrations/shadow_eval_logger.py b/litellm/integrations/shadow_eval_logger.py index 7f5d9fedd3d..0d059a89280 100644 --- a/litellm/integrations/shadow_eval_logger.py +++ b/litellm/integrations/shadow_eval_logger.py @@ -30,7 +30,11 @@ from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY from litellm.integrations.custom_logger import CustomLogger from litellm.integrations.websearch_interception.tools import is_web_search_tool_responses from litellm.litellm_core_utils.core_helpers import get_litellm_metadata_from_kwargs, independent_snapshot -from litellm.litellm_core_utils.internal_call_metadata import sanitized_forwardable_call_metadata +from litellm.litellm_core_utils.internal_call_metadata import ( + EvaluationBillingOwner, + evaluation_billing_context, + sanitized_forwardable_call_metadata, +) from litellm.litellm_core_utils.llm_judge import ( default_router_provider, extract_text_from_content, @@ -76,6 +80,7 @@ _EMPTY_METADATA: Final[Mapping[str, object]] = MappingProxyType({}) _CHAT_REQUEST_ADAPTER: Final = TypeAdapter(Mapping[str, object]) _CHAT_MESSAGES_ADAPTER: Final = TypeAdapter(tuple[Mapping[str, object], ...]) _MESSAGE_ITEMS_ADAPTER: Final = TypeAdapter(tuple[object, ...]) +_CREATOR_MODEL_BUDGET: Final = TypeAdapter[Mapping[str, object] | None](Mapping[str, object] | None) def _chat_messages(kwargs: Mapping[str, object]) -> tuple[Mapping[str, object], ...]: @@ -747,6 +752,7 @@ class ActiveShadowEvalJob(BaseModel): baseline_model: str | None = None shadow_percentage: float judge_model: str + created_by: str | None = None max_turns: int max_budget: float | None = None ends_at: datetime @@ -1082,22 +1088,27 @@ class ShadowEvalLogger(CustomLogger): if spend >= job.max_budget: self._record_funnel(job.id, "withheld") return - for arm_router in job.arm_router_names: - await self._run_shadow_arm( - prisma=prisma, - job=job, - arm_router=arm_router, - request_id=request_id, - messages=messages, - real_text=real_text, - real_model=real_model, - real_cost=real_cost, - real_classifier_cost=real_classifier_cost, - real_cache_hit=real_cache_hit, - control_tier=control_tier, - shadow_params=shadow_params, - parent_metadata=parent_metadata, - ) + owner: Final = await _evaluation_billing_owner(prisma, job.created_by) + if owner is None: + self._record_funnel(job.id, "withheld") + return + with evaluation_billing_context(owner): + for arm_router in job.arm_router_names: + await self._run_shadow_arm( + prisma=prisma, + job=job, + arm_router=arm_router, + request_id=request_id, + messages=messages, + real_text=real_text, + real_model=real_model, + real_cost=real_cost, + real_classifier_cost=real_classifier_cost, + real_cache_hit=real_cache_hit, + control_tier=control_tier, + shadow_params=shadow_params, + parent_metadata=parent_metadata, + ) async def _run_shadow_arm( self, @@ -1285,6 +1296,7 @@ class ShadowEvalLogger(CustomLogger): model=target_model, messages=[dict(m) for m in messages], # pyright: ignore[reportArgumentType] # snapshot of the SDK's own message dicts metadata=shadow_metadata, + stream=False, num_retries=0, fallbacks=[], **shadow_params, @@ -1348,6 +1360,7 @@ class ShadowEvalLogger(CustomLogger): judge_model, judge_messages, # pyright: ignore[reportArgumentType] # plain SDK message dicts team_id=_forwarded_team_id(parent_metadata), + stream=False, temperature=0, max_tokens=JUDGE_MAX_OUTPUT_TOKENS, response_format=PAIRWISE_JUDGE_RESPONSE_FORMAT, @@ -1380,3 +1393,32 @@ def _default_prisma_provider() -> "PrismaClient | None": except ImportError: return None return prisma_client + + +async def _evaluation_billing_owner(prisma: "PrismaClient", created_by: str | None) -> EvaluationBillingOwner | None: + from litellm.proxy.auth.auth_checks import get_user_object + from litellm.proxy.proxy_server import litellm_proxy_admin_name, proxy_logging_obj, user_api_key_cache + from litellm.types.proxy.auth.auth_checks import UserNotFoundError + + creator_id: Final = created_by or litellm_proxy_admin_name + try: + creator: Final = await get_user_object( + user_id=creator_id, + prisma_client=prisma, + user_api_key_cache=user_api_key_cache, + user_id_upsert=False, + proxy_logging_obj=proxy_logging_obj, + ) + except UserNotFoundError: + return None + except Exception as e: # noqa: BLE001 # optional evaluation work must not spend against unverifiable limits + verbose_logger.warning("shadow_eval: creator budget unavailable for %s: %s", creator_id, e) + return None + if creator is None: + return None + return EvaluationBillingOwner( + creator_id, + _CREATOR_MODEL_BUDGET.validate_python(creator.model_max_budget), + creator.max_budget, + creator.spend or 0.0, + ) diff --git a/litellm/litellm_core_utils/fallback_utils.py b/litellm/litellm_core_utils/fallback_utils.py index cc36f47b516..0e4179fe231 100644 --- a/litellm/litellm_core_utils/fallback_utils.py +++ b/litellm/litellm_core_utils/fallback_utils.py @@ -60,7 +60,7 @@ async def async_completion_with_fallbacks(**kwargs): completion_kwargs = filter_internal_params(completion_kwargs) response = await litellm.acompletion( - **completion_kwargs, + **{**completion_kwargs, "fallbacks": []}, model=model, litellm_logging_obj=litellm_logging_obj, ) diff --git a/litellm/litellm_core_utils/internal_call_metadata.py b/litellm/litellm_core_utils/internal_call_metadata.py index d844cbae367..9a00900461f 100644 --- a/litellm/litellm_core_utils/internal_call_metadata.py +++ b/litellm/litellm_core_utils/internal_call_metadata.py @@ -1,9 +1,7 @@ """Metadata a request forwards to the internal LLM sub-calls it triggers. -Internal features (the auto-router's classifier and embeddings, shadow eval's shadow and -judge calls) bill real provider spend that nobody typed a prompt for. That spend must land -on the same key/team/org/user as the request that caused it, so the sub-call carries the -caller's identity metadata, minus two things that must never be forwarded as-is: +Internal calls retain the caller's routing identity. Shadow evaluation receipts bill the +evaluation creator without changing that routing context. Two fields need special handling: * ``user_api_key_budget_reservation`` (and the reservation nested inside ``user_api_key_auth``) belongs to the parent completion. If a sub-call's cost callback @@ -17,16 +15,118 @@ caller's identity metadata, minus two things that must never be forwarded as-is: from __future__ import annotations -from collections.abc import Mapping +from collections.abc import Generator, Mapping +from contextlib import contextmanager +from contextvars import ContextVar +from dataclasses import dataclass from types import MappingProxyType from typing import Final +from pydantic import TypeAdapter + from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY, NON_INFERENCE_CALL_TYPES from litellm.litellm_core_utils.initialize_dynamic_callback_params import initialize_standard_callback_dynamic_params from litellm.types.utils import BACKGROUND_RESPONSE_COST_POLL_CALL_ORIGIN, InternalCallOrigin BUDGET_RESERVATION_METADATA_KEYS: Final = frozenset({"user_api_key_budget_reservation"}) + +@dataclass(frozen=True, slots=True) +class EvaluationBillingOwner: + user_id: str + user_model_max_budget: Mapping[str, object] | None = None + max_budget: float | None = None + spend: float = 0.0 + + +EVALUATION_BILLING_OWNER_KEY: Final = "_evaluation_billing_owner" +EVALUATION_BUDGET_RESERVATION_KEY: Final = "_evaluation_budget_reservation" +_EVALUATION_BILLING_OWNER: Final[ContextVar[EvaluationBillingOwner | None]] = ContextVar( + "evaluation_billing_owner", default=None +) +_BILLING_MAPPING: Final = TypeAdapter(Mapping[str, object]) +_BILLING_IDENTITY_FIELDS: Final = frozenset( + {"user_api_key", "user_api_end_user_max_budget", "team_id", "team_alias", "agent_id", "billing_agent_id"} +) + + +def get_evaluation_billing_owner() -> EvaluationBillingOwner | None: + return _EVALUATION_BILLING_OWNER.get() + + +@contextmanager +def evaluation_billing_context(owner: EvaluationBillingOwner | None) -> Generator[None]: + token: Final = _EVALUATION_BILLING_OWNER.set(owner) + try: + yield + finally: + _EVALUATION_BILLING_OWNER.reset(token) + + +def get_evaluation_billing_owner_from_kwargs(kwargs: Mapping[str, object]) -> EvaluationBillingOwner | None: + owner: Final = kwargs.get(EVALUATION_BILLING_OWNER_KEY) + return owner if isinstance(owner, EvaluationBillingOwner) else None + + +def _billing_mapping(value: object) -> Mapping[str, object]: + return _BILLING_MAPPING.validate_python(value) if isinstance(value, Mapping) else MappingProxyType({}) + + +def project_evaluation_billing_kwargs( + kwargs: Mapping[str, object], +) -> dict[str, object]: # mutable-ok: existing callback consumers require dictionaries + owner: Final = get_evaluation_billing_owner_from_kwargs(kwargs) + if owner is None: + return kwargs if isinstance(kwargs, dict) else dict(kwargs) + from litellm.proxy.spend_tracking.evaluation_budget import EvaluationAttempt + + handle: Final = kwargs.get(EVALUATION_BUDGET_RESERVATION_KEY) + reservation: Final = handle.total if isinstance(handle, EvaluationAttempt) else None + + def metadata(value: object) -> Mapping[str, object]: + return { + **{ + key: None if key.startswith("user_api_key_") or key in _BILLING_IDENTITY_FIELDS else item + for key, item in _billing_mapping(value).items() + }, + "user_api_key_user_id": owner.user_id, + "user_api_key_user_model_max_budget": owner.user_model_max_budget, + "user_api_key_budget_reservation": reservation, + "tags": [], + } + + def fields(value: object) -> Mapping[str, object]: + source: Final = _billing_mapping(value) + request: Final = _billing_mapping(source.get("proxy_server_request")) + return { + **source, + "user": owner.user_id, + "end_user": None, + "user_api_key_end_user_id": None, + "agent_id": None, + "billing_agent_id": None, + "request_tags": [], + "request_model_access_groups": (), + "metadata": metadata(source.get("metadata")), + **({"litellm_metadata": metadata(source["litellm_metadata"])} if source.get("litellm_metadata") else {}), + **( + {"proxy_server_request": {**request, "body": {**_billing_mapping(request["body"]), "user": None}}} + if isinstance(request.get("body"), Mapping) + else {} + ), + } + + return { + **fields(kwargs), + "litellm_params": fields(kwargs.get("litellm_params")), + **( + {"standard_logging_object": fields(kwargs["standard_logging_object"])} + if kwargs.get("standard_logging_object") is not None + else {} + ), + } + + MODEL_ACCESS_GROUP_METADATA_KEY: Final = "user_api_key_matched_model_access_groups" """Where auth records the model access groups that authorized the request, for the spend writer. diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 5a3a17f338c..4c1d5f8ec22 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -11,13 +11,14 @@ import sys import time import traceback from collections.abc import Awaitable, Callable, Iterator, Mapping, Sequence +from dataclasses import dataclass from datetime import datetime as dt_object from functools import lru_cache from types import MappingProxyType, TracebackType from typing import TYPE_CHECKING, Any, Final, Literal, Union, cast from httpx import Response -from pydantic import BaseModel, JsonValue +from pydantic import BaseModel, JsonValue, TypeAdapter import litellm from litellm import _custom_logger_compatible_callbacks_literal @@ -79,8 +80,13 @@ from litellm.litellm_core_utils.core_helpers import ( from litellm.litellm_core_utils.error_normalization import normalize_error from litellm.litellm_core_utils.get_litellm_params import get_litellm_params from litellm.litellm_core_utils.internal_call_metadata import ( + EVALUATION_BILLING_OWNER_KEY, + EVALUATION_BUDGET_RESERVATION_KEY, MODEL_ACCESS_GROUP_METADATA_KEY, + EvaluationBillingOwner, + get_evaluation_billing_owner, is_unbilled_non_inference_call, + project_evaluation_billing_kwargs, ) from litellm.litellm_core_utils.llm_cost_calc.guardrail_cost import ( cost_breakdown_with_guardrail, @@ -236,6 +242,7 @@ if TYPE_CHECKING: from litellm.litellm_core_utils.llm_cost_calc.utils import BilledTokenRates from litellm.llms.base_llm.passthrough.transformation import PassthroughStreamCollector from litellm.proxy.hooks.autorouter_baseline_cache import BaselineCacheContext, CapturedBaselineObservation + from litellm.proxy.spend_tracking.evaluation_budget import EvaluationAttempt try: from litellm_enterprise.enterprise_callbacks.callback_controls import ( EnterpriseCallbackControls, @@ -553,6 +560,22 @@ def _timestamp_seconds(moment: object) -> float | None: return None +@dataclass(slots=True) +class EvaluationBudgetInvocation: + reservation: "EvaluationAttempt | None" = None + + +def evaluation_logging_snapshot(logging_obj: "Logging") -> "Logging": + if not isinstance(logging_obj.evaluation_billing_owner, EvaluationBillingOwner): + return logging_obj + receipt: Final = copy.copy(logging_obj) + receipt.model_call_details.update(project_evaluation_billing_kwargs(receipt.model_call_details)) + receipt.litellm_params = TypeAdapter(dict[str, object]).validate_python( + receipt.model_call_details["litellm_params"] + ) + return receipt + + class Logging(LiteLLMLoggingBaseClass): global \ supabaseClient, \ @@ -618,6 +641,9 @@ class Logging(LiteLLMLoggingBaseClass): self.call_type = call_type self.litellm_call_id = litellm_call_id self.litellm_trace_id: str = litellm_trace_id if litellm_trace_id else str(uuid.uuid4()) + self.evaluation_billing_owner: Final[EvaluationBillingOwner | None] = get_evaluation_billing_owner() + self.evaluation_budget_reservation: EvaluationAttempt | None = None + self.evaluation_budget_invocation: EvaluationBudgetInvocation | None = None # Capture the pre-call *value* (not a contextvars.Token) so restoration works # even if this attempt's own logging ends up dispatched onto a different @@ -714,6 +740,8 @@ class Logging(LiteLLMLoggingBaseClass): "litellm_params": litellm_params, "applied_guardrails": applied_guardrails, "model": model, + EVALUATION_BILLING_OWNER_KEY: self.evaluation_billing_owner, + EVALUATION_BUDGET_RESERVATION_KEY: self.evaluation_budget_reservation, } # Set by proxy request handlers to defer spend-log fire until after @@ -945,6 +973,8 @@ class Logging(LiteLLMLoggingBaseClass): "standard_callback_dynamic_params": self.standard_callback_dynamic_params, **self.optional_params, **additional_params, + EVALUATION_BILLING_OWNER_KEY: self.evaluation_billing_owner, + EVALUATION_BUDGET_RESERVATION_KEY: self.evaluation_budget_reservation, } ) @@ -2197,7 +2227,7 @@ class Logging(LiteLLMLoggingBaseClass): return executor.submit( - self.success_handler, + evaluation_logging_snapshot(self).success_handler, result, start_time=start_time, end_time=end_time, @@ -2216,6 +2246,22 @@ class Logging(LiteLLMLoggingBaseClass): if isinstance(usage, Usage): self.record_partial_usage_for_failure(usage, self._response_cost_calculator(result=assembled) or 0.0) + def recover_failure_cost(self, result: object) -> float: + if isinstance(result, ModelResponse): + self.record_assembled_response_for_failure(result) + elif isinstance(result, ResponsesAPIResponse): + self.record_assembled_response_for_failure(self._translate_responses_api_response_to_model_response(result)) + self.model_call_details["response_cost"] = self._response_cost_calculator(result=result) + elif result is not None and self.call_type == CallTypes.anthropic_messages.value: + self.record_assembled_response_for_failure(self._handle_anthropic_messages_response_logging(result)) + return TypeAdapter(float).validate_python(self.model_call_details.get("response_cost") or 0.0) + + async def _settle_evaluation_budget(self) -> None: + if self.evaluation_budget_reservation is not None: + await self.evaluation_budget_reservation.settle( + TypeAdapter(float).validate_python(self.model_call_details.get("response_cost") or 0.0) + ) + async def dispatch_failure_handlers( self, exception: Exception, @@ -2244,7 +2290,7 @@ class Logging(LiteLLMLoggingBaseClass): await self.async_failure_handler(exception, traceback_exception) finally: if self._should_run_sync_failure_callbacks_for_async_calls(): - executor.submit(self.failure_handler, exception, traceback_exception) + executor.submit(evaluation_logging_snapshot(self).failure_handler, exception, traceback_exception) def should_run_logging( self, @@ -2505,6 +2551,8 @@ class Logging(LiteLLMLoggingBaseClass): standard_logging_object: StandardLoggingPayload | None = None, build_logging_payload: bool = True, ): + if self.evaluation_billing_owner is not None: + self.model_call_details.update(project_evaluation_billing_kwargs(self.model_call_details)) try: if start_time is None: start_time = self.start_time @@ -3395,6 +3443,9 @@ class Logging(LiteLLMLoggingBaseClass): # print standard logging payload if (standard_logging_payload := self.model_call_details.get("standard_logging_object")) is not None: emit_standard_logging_payload(standard_logging_payload) + if not self.stream or "async_complete_streaming_response" in self.model_call_details: + await self._settle_evaluation_budget() + callbacks: Final = self.get_combined_callback_list( dynamic_success_callbacks=self.dynamic_async_success_callbacks, global_callbacks=litellm._async_success_callback, @@ -3605,6 +3656,8 @@ class Logging(LiteLLMLoggingBaseClass): verbose_logger.debug("Error in _handle_callback_failure: %s", e) def _failure_handler_helper_fn(self, exception, traceback_exception, start_time=None, end_time=None): + if self.evaluation_billing_owner is not None and hasattr(self, "model_call_details"): + self.model_call_details.update(project_evaluation_billing_kwargs(self.model_call_details)) if start_time is None: start_time = self.start_time if end_time is None: @@ -3639,6 +3692,9 @@ class Logging(LiteLLMLoggingBaseClass): metadata: Final = self.model_call_details["litellm_params"].get("metadata", {}) or {} metadata.update(exception.headers) + if self.evaluation_billing_owner is not None: + self.model_call_details.update(project_evaluation_billing_kwargs(self.model_call_details)) + ## STANDARDIZED LOGGING PAYLOAD self.model_call_details["standard_logging_object"] = get_standard_logging_object_payload( @@ -3920,6 +3976,7 @@ class Logging(LiteLLMLoggingBaseClass): start_time=start_time, end_time=end_time, ) + await self._settle_evaluation_budget() callbacks: Final = self.get_combined_callback_list( dynamic_success_callbacks=self.dynamic_async_failure_callbacks, @@ -4002,7 +4059,7 @@ class Logging(LiteLLMLoggingBaseClass): return executor.submit( - self.success_handler, + evaluation_logging_snapshot(self).success_handler, result, start_time, end_time, @@ -6537,8 +6594,12 @@ def get_standard_logging_object_payload( _model_id: Final = metadata.get("model_info", {}).get("id", "") _model_group: Final = metadata.get("model_group", "") - request_tags: Final = StandardLoggingPayloadSetup._get_request_tags( - litellm_params=litellm_params, proxy_server_request=proxy_server_request + request_tags: Final = ( + [] + if logging_obj.evaluation_billing_owner is not None + else StandardLoggingPayloadSetup._get_request_tags( + litellm_params=litellm_params, proxy_server_request=proxy_server_request + ) ) request_model_access_groups: Final = request_model_access_groups_from_litellm_params(litellm_params) diff --git a/litellm/litellm_core_utils/llm_judge.py b/litellm/litellm_core_utils/llm_judge.py index ed3b89dd420..041e61a388a 100644 --- a/litellm/litellm_core_utils/llm_judge.py +++ b/litellm/litellm_core_utils/llm_judge.py @@ -118,7 +118,7 @@ async def judge_acompletion( ) -> ModelResponse: """Dispatch a judge call through the proxy's router when the judge model is a configured deployment (DB-stored credentials work), through the SDK for - provider-qualified public names. The router path never retries or falls back: + provider-qualified public names. Judge calls never retry or fall back: a failed judge call is the caller's counted failure, not a spend multiplier. Sampling preferences are advisory: models that removed sampling params (e.g. claude-sonnet-5) drop them instead of rejecting the judge call. @@ -135,4 +135,6 @@ async def judge_acompletion( drop_params=True, **params, ) - return await litellm.acompletion(model=judge_model, messages=messages, num_retries=0, drop_params=True, **params) + return await litellm.acompletion( + model=judge_model, messages=messages, num_retries=0, fallbacks=[], drop_params=True, **params + ) diff --git a/litellm/llms/anthropic/pass_through/messages/handler.py b/litellm/llms/anthropic/pass_through/messages/handler.py index 3068a7eb3ab..d8457c72dde 100644 --- a/litellm/llms/anthropic/pass_through/messages/handler.py +++ b/litellm/llms/anthropic/pass_through/messages/handler.py @@ -7,10 +7,12 @@ import asyncio import contextvars -from collections.abc import AsyncIterator, Coroutine, Iterator +from collections.abc import AsyncIterator, Coroutine, Iterator, Mapping from functools import partial from typing import Any, Final, cast +from pydantic import TypeAdapter + import litellm from litellm.litellm_core_utils.exception_mapping_utils import exception_type from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj @@ -106,6 +108,32 @@ def _deployment_supports_cache_control_ttl(model_info: object) -> bool: return isinstance(model_info, dict) and model_info.get("cache_control_ttl") is True +def get_anthropic_messages_provider_config( + model: str, custom_llm_provider: str | None, model_info: object +) -> BaseAnthropicMessagesConfig | None: + from litellm.types.utils import LlmProviders + + config: Final = ( + ProviderConfigManager.get_provider_anthropic_messages_config(model, LlmProviders(custom_llm_provider)) + if custom_llm_provider is not None and custom_llm_provider in (provider.value for provider in LlmProviders) + else None + ) + if config is not None or not _deployment_passes_through_anthropic_messages(model_info): + return config + from litellm.llms.openai_like.messages.transformation import OpenAILikeAnthropicMessagesConfig + + return OpenAILikeAnthropicMessagesConfig(cache_control_ttl=_deployment_supports_cache_control_ttl(model_info)) + + +def anthropic_messages_uses_bridge(model: str, request: Mapping[str, object]) -> bool: + resolved_model, provider, _, _ = litellm.get_llm_provider( + model=model, + custom_llm_provider=TypeAdapter[str | None](str | None).validate_python(request.get("custom_llm_provider")), + api_base=TypeAdapter[str | None](str | None).validate_python(request.get("api_base")), + ) + return get_anthropic_messages_provider_config(resolved_model, provider, request.get("model_info")) is None + + ####### ENVIRONMENT VARIABLES ################### # Initialize any necessary instances or variables here base_llm_http_handler = BaseLLMHTTPHandler() @@ -452,8 +480,6 @@ def anthropic_messages_handler( Args: container: Container config with skills for code execution """ - from litellm.types.utils import LlmProviders - # Sanitize empty text blocks so the sync entry point # (litellm.messages.create -> anthropic_messages_handler) gets the same # protection as the async wrapper. The async wrapper already sanitized and @@ -561,23 +587,9 @@ def anthropic_messages_handler( **kwargs, ) - anthropic_messages_provider_config: BaseAnthropicMessagesConfig | None = None - - if custom_llm_provider is not None and custom_llm_provider in [provider.value for provider in LlmProviders]: - anthropic_messages_provider_config = ProviderConfigManager.get_provider_anthropic_messages_config( - model=model, - provider=litellm.LlmProviders(custom_llm_provider), - ) - if anthropic_messages_provider_config is None and _deployment_passes_through_anthropic_messages( - kwargs.get("model_info") - ): - from litellm.llms.openai_like.messages.transformation import ( - OpenAILikeAnthropicMessagesConfig, - ) - - anthropic_messages_provider_config = OpenAILikeAnthropicMessagesConfig( - cache_control_ttl=_deployment_supports_cache_control_ttl(kwargs.get("model_info")), - ) + anthropic_messages_provider_config: Final = get_anthropic_messages_provider_config( + model, custom_llm_provider, TypeAdapter(object).validate_python(kwargs.get("model_info")) + ) if anthropic_messages_provider_config is None: # Route to Responses API for OpenAI / Azure, chat/completions for everything else. if kwargs.get("compaction") is None and _should_route_to_responses_api( diff --git a/litellm/main.py b/litellm/main.py index a818213b861..0c8dc58d9d5 100644 --- a/litellm/main.py +++ b/litellm/main.py @@ -641,9 +641,11 @@ async def acompletion( api_base=kwargs.get("api_base") or base_url, ) - fallbacks = fallbacks or litellm.model_fallbacks - if fallbacks is not None: - response = await async_completion_with_fallbacks(**completion_kwargs, kwargs={"fallbacks": fallbacks, **kwargs}) + resolved_fallbacks: Final = litellm.model_fallbacks if fallbacks is None else fallbacks + if resolved_fallbacks: + response = await async_completion_with_fallbacks( + **completion_kwargs, kwargs={**kwargs, "fallbacks": resolved_fallbacks} + ) if response is None: raise Exception( "No response from fallbacks. Got none. Turn on `litellm.set_verbose=True` to see more details." @@ -5458,10 +5460,10 @@ def completion( elif num_retries is not None: max_retries = num_retries logging: Final[LiteLLMLoggingObj] = cast(LiteLLMLoggingObj, litellm_logging_obj) - fallbacks = fallbacks or litellm.model_fallbacks - if fallbacks is not None: + resolved_fallbacks: Final = litellm.model_fallbacks if fallbacks is None else fallbacks + if resolved_fallbacks: return completion_with_fallbacks( # pyright: ignore[reportReturnType] # fallback runner is untyped; resolves to ModelResponse|CustomStreamWrapper at runtime - **args + **{**args, "kwargs": {**kwargs, "fallbacks": resolved_fallbacks}} ) if model_list is not None: deployments: Final = [m["litellm_params"] for m in model_list if m["model_name"] == model] diff --git a/litellm/proxy/auth/user_api_key_auth.py b/litellm/proxy/auth/user_api_key_auth.py index 5395d817e33..ea71a46db22 100644 --- a/litellm/proxy/auth/user_api_key_auth.py +++ b/litellm/proxy/auth/user_api_key_auth.py @@ -19,6 +19,7 @@ import fastapi import orjson from fastapi import HTTPException, Request, WebSocket, status from fastapi.security.api_key import APIKeyHeader +from pydantic import TypeAdapter from starlette.exceptions import WebSocketException import litellm @@ -39,6 +40,7 @@ from litellm.integrations.otel.model.config import is_otel_v2_enabled from litellm.integrations.otel.runtime import phase_event, phase_span, seed_request_identity from litellm.litellm_core_utils.dd_tracing import tracer from litellm.litellm_core_utils.dot_notation_indexing import get_nested_value +from litellm.litellm_core_utils.internal_call_metadata import get_evaluation_billing_owner from litellm.proxy._types import * from litellm.proxy.agent_endpoints.auth.agent_caller import agent_caller_from_headers from litellm.proxy.auth.auth_checks import ( @@ -3224,13 +3226,17 @@ async def _reserve_budget_after_common_checks( request: Request | None = None, ) -> None: user_api_key_auth_obj.budget_reservation = None - if not skip_budget_checks and general_settings.get("disable_budget_reservation") is not True: + if ( + not skip_budget_checks + and general_settings.get("disable_budget_reservation") is not True + and get_evaluation_billing_owner() is None + ): from litellm.proxy.spend_tracking.budget_reservation import ( reserve_budget_for_request, ) user_api_key_auth_obj.budget_reservation = await reserve_budget_for_request( - request_body=request_data, + request_body=TypeAdapter(dict[str, object]).validate_python(request_data), route=route, llm_router=llm_router, valid_token=user_api_key_auth_obj, diff --git a/litellm/proxy/hooks/model_max_budget_limiter.py b/litellm/proxy/hooks/model_max_budget_limiter.py index d019271d404..5f7be2da0bf 100644 --- a/litellm/proxy/hooks/model_max_budget_limiter.py +++ b/litellm/proxy/hooks/model_max_budget_limiter.py @@ -470,7 +470,13 @@ class _PROXY_VirtualKeyModelMaxBudgetLimiter(RouterBudgetLimiting): model=model, resolved=resolved, ) - current_spend: Final = _as_spend(await self._cached_spend(spend_key)) + from litellm.proxy.spend_tracking.evaluation_budget import model_budget_spend + + current_spend: Final = ( + await model_budget_spend(self.dual_cache, spend_key) + if entity_type == Litellm_EntityType.USER + else _as_spend(await self._cached_spend(spend_key)) + ) if legacy_spend_key is None or legacy_spend_key == spend_key: return current_spend return current_spend + _as_spend(await self._cached_spend(legacy_spend_key)) @@ -499,6 +505,14 @@ class _PROXY_VirtualKeyModelMaxBudgetLimiter(RouterBudgetLimiting): Example: key=sk-1234567890, model=gpt-4o, max_budget=100, time_period=1d """ + from litellm.litellm_core_utils.internal_call_metadata import EVALUATION_BUDGET_RESERVATION_KEY + from litellm.proxy.spend_tracking.evaluation_budget import EvaluationAttempt + + if ( + isinstance(attempt := kwargs.get(EVALUATION_BUDGET_RESERVATION_KEY), EvaluationAttempt) + and attempt.model is not None + ): + return verbose_proxy_logger.debug("in RouterBudgetLimiting.async_log_success_event") standard_logging_payload: Final[StandardLoggingPayload | None] = kwargs.get("standard_logging_object", None) if standard_logging_payload is None: diff --git a/litellm/proxy/hooks/proxy_track_cost_callback.py b/litellm/proxy/hooks/proxy_track_cost_callback.py index f937b439042..06d539ceda1 100644 --- a/litellm/proxy/hooks/proxy_track_cost_callback.py +++ b/litellm/proxy/hooks/proxy_track_cost_callback.py @@ -4,6 +4,8 @@ from collections.abc import Callable, Mapping, Sequence from datetime import datetime from typing import TYPE_CHECKING, Any, Final, Protocol, cast +from pydantic import TypeAdapter + import litellm from litellm._logging import verbose_proxy_logger from litellm.batches.batch_utils import batch_cost_is_final @@ -15,7 +17,14 @@ from litellm.litellm_core_utils.core_helpers import ( get_litellm_metadata_from_kwargs, get_metadata_variable_name_from_kwargs, ) -from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup +from litellm.litellm_core_utils.internal_call_metadata import ( + EVALUATION_BILLING_OWNER_KEY, + EVALUATION_BUDGET_RESERVATION_KEY, + get_evaluation_billing_owner, + get_evaluation_billing_owner_from_kwargs, + project_evaluation_billing_kwargs, +) +from litellm.litellm_core_utils.litellm_logging import Logging, StandardLoggingPayloadSetup from litellm.litellm_core_utils.llm_cost_calc.guardrail_cost import guardrail_information_cost from litellm.proxy._types import UserAPIKeyAuth from litellm.proxy.auth.auth_checks import ( @@ -273,19 +282,34 @@ class _ProxyDBLogger(CustomLogger): recovered_response_cost: Final = recovered_stream_cost + guardrail_information_cost( existing_metadata.get("standard_logging_guardrail_information") ) + logger: Final = _litellm_logging_obj if isinstance(_litellm_logging_obj, Logging) else None + request_receipt: Final = TypeAdapter(dict[str, object]).validate_python(request_data) + owner: Final = ( + get_evaluation_billing_owner_from_kwargs(request_receipt) + or (logger.evaluation_billing_owner if logger is not None else None) + or get_evaluation_billing_owner() + ) + billing_auth: Final = UserAPIKeyAuth(user_id=owner.user_id) if owner is not None else user_api_key_dict + receipt: Final = project_evaluation_billing_kwargs( + { + **request_receipt, + EVALUATION_BILLING_OWNER_KEY: owner, + EVALUATION_BUDGET_RESERVATION_KEY: logger.evaluation_budget_reservation if logger is not None else None, + } + ) await self._spend_writer().update_database( - token=LiteLLMProxyRequestSetup.get_logged_api_key(user_api_key_dict), + token=LiteLLMProxyRequestSetup.get_logged_api_key(billing_auth), response_cost=recovered_response_cost, - user_id=user_api_key_dict.user_id, - end_user_id=user_api_key_dict.end_user_id, - team_id=user_api_key_dict.team_id, - kwargs=request_data, + user_id=billing_auth.user_id, + end_user_id=billing_auth.end_user_id, + team_id=billing_auth.team_id, + kwargs=receipt, completion_response=original_exception, start_time=actual_start_time, end_time=datetime.now(), - org_id=user_api_key_dict.org_id, - project_id=user_api_key_dict.project_id, + org_id=billing_auth.org_id, + project_id=billing_auth.project_id, ) async def _PROXY_track_cost_callback( diff --git a/litellm/proxy/management_endpoints/auto_router_endpoints.py b/litellm/proxy/management_endpoints/auto_router_endpoints.py index 33fb069afbd..3a8aca01dac 100644 --- a/litellm/proxy/management_endpoints/auto_router_endpoints.py +++ b/litellm/proxy/management_endpoints/auto_router_endpoints.py @@ -1684,10 +1684,10 @@ async def start_shadow_eval( eval spend, the shadow and judge calls' own cost, reaches max_budget dollars, the job's window ends, or the job is stopped, so one target running out of budget does not end sampling for the others; sampling changes propagate to pods within about 10 - seconds. Shadow and judge calls bill to the sampled request's own identity but are + seconds. Shadow and judge calls bill to the admin who started the job and are excluded from request counts and auto-router adoption metrics. """ - from litellm.proxy.proxy_server import llm_router, prisma_client + from litellm.proxy.proxy_server import litellm_proxy_admin_name, llm_router, prisma_client _require_admin_writer(user_api_key_dict, "start a shadow eval") if prisma_client is None: @@ -1805,7 +1805,7 @@ async def start_shadow_eval( "shadow_percentage": data.shadow_percentage, "max_turns": SHADOW_EVAL_TURN_VALVE, "max_budget": data.max_budget, - "created_by": user_api_key_dict.user_id, + "created_by": user_api_key_dict.user_id or litellm_proxy_admin_name, "created_at": now, "ends_at": ends_at, } diff --git a/litellm/proxy/native_compaction.py b/litellm/proxy/native_compaction.py index fd27e7fbbdc..f30c480c677 100644 --- a/litellm/proxy/native_compaction.py +++ b/litellm/proxy/native_compaction.py @@ -13,6 +13,10 @@ from litellm.litellm_core_utils.initialize_dynamic_callback_params import ( inherit_message_logging_privacy, initialize_standard_callback_dynamic_params, ) +from litellm.litellm_core_utils.internal_call_metadata import ( + evaluation_billing_context, + get_evaluation_billing_owner, +) from litellm.llms.custom_httpx.asgi_handler import get_async_asgi_client from litellm.proxy.litellm_pre_call_utils import UNTRUSTED_REQUEST_HEADER_CONTROL_FIELDS from litellm.router_strategy.complexity_router.context_compaction import ( @@ -40,6 +44,7 @@ async def with_proxy_compaction_executor(call: Awaitable[_ResultT], request: Req protocol: Literal["chat", "messages"], payload: Mapping[str, object], parent_model: str | None = None ) -> Mapping[str, object]: logging_disabled: Final = initialize_standard_callback_dynamic_params().get("turn_off_message_logging") is True + billing_owner: Final = get_evaluation_billing_owner() async def dispatch() -> Mapping[str, object]: scope: Final = _JSON_OBJECT.validate_python(request.scope) @@ -55,6 +60,7 @@ async def with_proxy_compaction_executor(call: Awaitable[_ResultT], request: Req with ( native_compaction_call(parent_model, str(payload["model"])), inherit_message_logging_privacy(logging_disabled), + evaluation_billing_context(billing_owner), ): with get_async_asgi_client( app=_ASGI_APP.validate_python(scope["app"]), diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 0e88ff9c13f..e15414de877 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -3373,15 +3373,21 @@ async def _reconcile_budget_reservation_for_counter_update( budget_reservation: dict | None, response_cost: float | None, ) -> _ReservationCounterUpdate: - if budget_reservation is None or budget_reservation.get("finalized") is True: - return _ReservationCounterUpdate() - from litellm.proxy.spend_tracking.budget_reservation import ( get_reserved_counter_keys, invalidate_budget_reservation_counters, reconcile_budget_reservation, ) + if budget_reservation is not None and budget_reservation.get("externally_settled") is True: + return _ReservationCounterUpdate( + reserved_counter_keys=TypeAdapter(frozenset[str]).validate_python( + get_reserved_counter_keys(budget_reservation) + ) + ) + if budget_reservation is None or budget_reservation.get("finalized") is True: + return _ReservationCounterUpdate() + reserved_counter_keys: Final = get_reserved_counter_keys(budget_reservation=budget_reservation) try: pending: Final = await reconcile_budget_reservation( diff --git a/litellm/proxy/spend_tracking/budget_reservation.py b/litellm/proxy/spend_tracking/budget_reservation.py index 2dd1c9419f1..3c224e27936 100644 --- a/litellm/proxy/spend_tracking/budget_reservation.py +++ b/litellm/proxy/spend_tracking/budget_reservation.py @@ -111,7 +111,11 @@ def get_reserved_counter_keys(budget_reservation: dict | None) -> set: _lease_renewals: Final[set[asyncio.Task[None]]] = set() # mutable-ok: asyncio only weak-refs pending tasks -def _start_reservation_lease_renewal(budget_reservation: Mapping[str, object], counter_keys: frozenset[str]) -> None: +def _start_reservation_lease_renewal( + budget_reservation: Mapping[str, object], + counter_keys: frozenset[str], + request_task: asyncio.Task[object] | None = None, +) -> None: """A reservation lives inside spend counter keys that expire on their Redis TTL. Renew the TTL while the request is in flight so a request longer than the TTL does not drop its reservation and admit concurrent requests against the DB floor on any worker.""" @@ -124,7 +128,7 @@ def _start_reservation_lease_renewal(budget_reservation: Mapping[str, object], c budget_reservation=budget_reservation, counter_keys=counter_keys, interval=spend_counter_cache.redis_cache.default_ttl / 2, - request_task=asyncio.current_task(), + request_task=request_task or asyncio.current_task(), ) ) _lease_renewals.add(task) @@ -144,7 +148,11 @@ async def _renew_reservation_lease( deadline: Final = time.monotonic() + litellm.request_timeout while time.monotonic() < deadline: await asyncio.sleep(interval) - if budget_reservation.get("finalized") is True or (request_task is not None and request_task.done()): + if ( + budget_reservation.get("finalized") is True + or budget_reservation.get("externally_settled") is True + or (request_task is not None and request_task.done()) + ): return for counter_key in counter_keys: await refresh_spend_counter_ttl(counter_key=counter_key) @@ -248,7 +256,7 @@ def _is_unbilled_route(route: str) -> bool: async def reserve_budget_for_request( - request_body: dict, + request_body: dict[str, object], # mutable-ok: existing reservation helpers accept dictionaries route: str, llm_router: Router | None, valid_token: UserAPIKeyAuth | None, @@ -262,7 +270,8 @@ async def reserve_budget_for_request( apply_user_budget_to_team_keys: bool = False, fail_closed_budget_enforcement: bool = False, raw_body: bytes | None = None, -) -> dict | None: + request_task: asyncio.Task[object] | None = None, +) -> dict[str, object] | None: # mutable-ok: shared reservation is finalized by the spend writer if valid_token is None or not RouteChecks.is_llm_api_route(route=route): return None if _is_unbilled_route(route): @@ -334,7 +343,7 @@ async def reserve_budget_for_request( llm_router=llm_router, input_token_counts=input_token_counts, ) - budget_reservation: Final = { + budget_reservation: Final[dict[str, object]] = { # mutable-ok: shared finalization state "reserved_cost": reservation_cost, "entries": applied_entries, "finalized": False, @@ -345,6 +354,7 @@ async def reserve_budget_for_request( _start_reservation_lease_renewal( budget_reservation=budget_reservation, counter_keys=frozenset(get_reserved_counter_keys(budget_reservation=budget_reservation)), + request_task=request_task, ) return budget_reservation @@ -357,8 +367,13 @@ async def reconcile_budget_reservation( ) -> tuple[PendingSpendIncrement, ...]: """Settle every reserved counter on ``actual_cost``. With ``apply_consistent`` False the adjustments for counters that still hold the reservation are returned instead of written, so the caller can pipeline them with - its own increments and then call ``stamp_budget_reservation_actual_cost``.""" - if not budget_reservation or budget_reservation.get("finalized") is True: + its own increments and then call ``stamp_budget_reservation_actual_cost``. External owners settle independently; + their entries remain available for the spend writer to skip already charged counters.""" + if ( + not budget_reservation + or budget_reservation.get("finalized") is True + or budget_reservation.get("externally_settled") is True + ): return () reserved_cost: Final = float(budget_reservation.get("reserved_cost") or 0.0) @@ -377,7 +392,7 @@ async def reconcile_budget_reservation( def stamp_budget_reservation_actual_cost(budget_reservation: dict | None, actual_cost: float | None) -> None: """Record that every reserved counter now holds ``actual_cost``, once the adjustments handed back by ``reconcile_budget_reservation(apply_consistent=False)`` have been written.""" - if not budget_reservation: + if not budget_reservation or budget_reservation.get("externally_settled") is True: return reserved_cost: Final = float(budget_reservation.get("reserved_cost") or 0.0) actual: Final = float(actual_cost or 0.0) @@ -1302,7 +1317,7 @@ def _coerce_datetime(value: object) -> datetime | None: def estimate_request_max_cost( - request_body: dict, + request_body: dict[str, object], # mutable-ok: existing cost helpers accept dictionaries route: str, llm_router: Router | None, input_token_counts: Mapping[str, int] | None = None, @@ -1324,7 +1339,7 @@ def estimate_request_max_cost( def estimate_request_input_cost( - request_body: dict, + request_body: dict[str, object], # mutable-ok: existing cost helpers accept dictionaries route: str, llm_router: Router | None, input_token_counts: Mapping[str, int] | None = None, diff --git a/litellm/proxy/spend_tracking/evaluation_budget.py b/litellm/proxy/spend_tracking/evaluation_budget.py new file mode 100644 index 00000000000..b64a897d516 --- /dev/null +++ b/litellm/proxy/spend_tracking/evaluation_budget.py @@ -0,0 +1,383 @@ +from __future__ import annotations + +import asyncio +import math +import time +import uuid +from collections.abc import Awaitable, Mapping, Sequence +from dataclasses import dataclass, field +from functools import lru_cache +from itertools import product +from string import ascii_letters, digits +from threading import Lock +from typing import Final, Literal, Protocol, runtime_checkable + +from pydantic import ConfigDict, TypeAdapter +from redis.crc import key_slot + +import litellm +from litellm._internal_context import with_service_target +from litellm._logging import verbose_proxy_logger +from litellm.caching.caching import DualCache +from litellm.litellm_core_utils.duration_parser import duration_in_seconds +from litellm.litellm_core_utils.internal_call_metadata import EvaluationBillingOwner +from litellm.proxy._types import Litellm_EntityType, LiteLLM_UserTable, UserAPIKeyAuth +from litellm.proxy.hooks.model_max_budget_limiter import ( + model_budget_spend_cache_key, + model_budget_start_time_cache_key, + resolve_model_budget, +) +from litellm.proxy.spend_tracking.budget_reservation import ( + estimate_request_input_cost, + estimate_request_max_cost, + reconcile_budget_reservation, # pyright: ignore[reportUnknownVariableType] # legacy reservation entries are untyped + reserve_budget_for_request, +) +from litellm.router_utils.common_utils import resolve_model_group_alias +from litellm.types.utils import API_ROUTE_TO_CALL_TYPES, CallTypes + +_NUMBER: Final = TypeAdapter(float) +_MAPPING: Final = TypeAdapter(Mapping[str, object]) +_REQUEST: Final = TypeAdapter(dict[str, object]) +_HOLDS: Final = TypeAdapter(Mapping[str, float]) +_LEASE_SECONDS: Final = 60 +_LOCAL_LOCK: Final = Lock() +_LEASES: Final[set[asyncio.Task[None]]] = set() # mutable-ok: asyncio weakly references pending tasks +_HOLD_SCRIPT: Final = """ +local actual = tonumber(redis.call('GET', KEYS[2]) or '0') +if not actual then return redis.error_reply('Invalid model budget spend') end +local clock = redis.call('TIME') +local now = tonumber(clock[1]) + tonumber(clock[2]) / 1000000 +local total = 0 +for _, member in ipairs(redis.call('ZRANGEBYSCORE', KEYS[1], '(' .. now, '+inf')) do + if member ~= ARGV[2] then total = total + tonumber(string.match(member, ':([^:]+)$')) end +end +if ARGV[1] == 'renew' then + local expiry = redis.call('ZSCORE', KEYS[1], ARGV[2]) + if not expiry or tonumber(expiry) <= now then return redis.error_reply('Evaluation reservation expired') end +end +if ARGV[1] == 'reserve' or ARGV[1] == 'renew' then + total = total + tonumber(string.match(ARGV[2], ':([^:]+)$')) +end +if ARGV[1] == 'reserve' and ARGV[6] ~= '' then + local estimate = tonumber(string.match(ARGV[2], ':([^:]+)$')) + if actual + total > tonumber(ARGV[6]) or actual + total - estimate >= tonumber(ARGV[6]) then + return tostring(actual + total) + end +end +if ARGV[1] == 'settle' and tonumber(ARGV[4]) > 0 then + actual = tonumber(redis.call('INCRBYFLOAT', KEYS[2], ARGV[4])) + redis.call('EXPIRE', KEYS[2], ARGV[5]) +end +redis.call('ZREMRANGEBYSCORE', KEYS[1], '-inf', now) +if ARGV[1] == 'reserve' or ARGV[1] == 'renew' then + redis.call('ZADD', KEYS[1], now + tonumber(ARGV[3]), ARGV[2]) + redis.call('EXPIRE', KEYS[1], ARGV[3]) +elseif ARGV[1] == 'settle' then redis.call('ZREM', KEYS[1], ARGV[2]) end +return tostring(actual + total) +""" + + +@runtime_checkable +class _NumericCache(Protocol): + def get_cache(self, key: str) -> object: ... + def set_cache(self, key: str, value: object, *, ttl: int) -> object: ... + def delete_cache(self, key: str) -> object: ... + def increment_cache(self, key: str, value: float, *, ttl: int, refresh_ttl: bool = False) -> float: ... + + +@runtime_checkable +class _Script(Protocol): + async def __call__(self, *, keys: Sequence[str], args: Sequence[str | int | float]) -> object: ... + + +_LOCAL: Final = TypeAdapter(_NumericCache, config=ConfigDict(arbitrary_types_allowed=True)) +_SCRIPT: Final = TypeAdapter(_Script, config=ConfigDict(arbitrary_types_allowed=True)) + + +@lru_cache(maxsize=4096) +def _model_hold_key(effective_key: str) -> str: + prefix: Final = f"{effective_key}:evaluation_holds:" + tagged: Final = f"{prefix}{{{effective_key}}}" + slot: Final = key_slot(effective_key.encode()) + if key_slot(tagged.encode()) == slot: + return tagged + candidates: Final = (prefix + "".join(chars) for chars in product(ascii_letters + digits + "-_", repeat=3)) + return next(candidate for candidate in candidates if key_slot(candidate.encode()) == slot) + + +@with_service_target("model_budgets") +async def model_budget_spend( + cache: DualCache, + spend_key: str, + *, + operation: Literal["read", "reserve", "renew", "settle"] = "read", + member: str = "", + adjustment: float = 0.0, + ttl: int = 1, + limit: float | None = None, +) -> float: + if cache.redis_cache is not None: + key: Final = cache.redis_cache.check_and_fix_namespace(spend_key) + script: Final = _SCRIPT.validate_python(cache.redis_cache.async_register_script(_HOLD_SCRIPT)) + return _NUMBER.validate_python( + await script( + keys=(_model_hold_key(key), key), + args=(operation, member, _LEASE_SECONDS, adjustment, ttl, "" if limit is None else limit), + ) + ) + local: Final = _LOCAL.validate_python(cache.in_memory_cache) + hold_key: Final = _model_hold_key(spend_key) + with _LOCAL_LOCK: + actual: Final = _NUMBER.validate_python(local.get_cache(spend_key) or 0.0) + now: Final = cache.in_memory_cache._clock() # pyright: ignore[reportPrivateUsage] # lease expiry uses the cache's injected clock + active: Final = { + token: expiry + for token, expiry in _HOLDS.validate_python(local.get_cache(hold_key) or {}).items() + if expiry > now + } + if operation == "renew" and member not in active: + raise RuntimeError("Evaluation reservation expired") + updated: Final = { + **{token: expiry for token, expiry in active.items() if token != member}, + **({member: now + _LEASE_SECONDS} if operation in ("reserve", "renew") else {}), + } + held: Final = sum(float(token.rsplit(":", 1)[1]) for token in updated) + proposed: Final = actual + held + if ( + operation == "reserve" + and limit is not None + and (proposed > limit or proposed - float(member.rsplit(":", 1)[1]) >= limit) + ): + return proposed + settled: Final = ( + local.increment_cache(spend_key, adjustment, ttl=ttl, refresh_ttl=True) + if operation == "settle" and adjustment + else actual + ) + local.delete_cache(hold_key) + if updated: + local.set_cache(hold_key, updated, ttl=_LEASE_SECONDS) + return settled + held + + +async def _complete(operation: Awaitable[EvaluationAttempt | None]) -> EvaluationAttempt | None: + task: Final = asyncio.ensure_future(operation) + try: + return await asyncio.shield(task) + except asyncio.CancelledError: + drained: Final = asyncio.gather(task, return_exceptions=True) + while not drained.done(): + try: + await asyncio.shield(drained) + except asyncio.CancelledError: + continue + if not task.cancelled() and task.exception() is None and (attempt := task.result()) is not None: + await attempt.settle(0.0) + raise + + +@dataclass(frozen=True, slots=True) +class _ModelBudget: + spend_key: str + start_key: str + duration: int + limit: float + + +@with_service_target("model_budgets") +async def _model_window(cache: DualCache, model: _ModelBudget) -> int: + if cache.redis_cache is not None: + script: Final = _SCRIPT.validate_python( + cache.redis_cache.async_register_script( + "local clock = redis.call('TIME'); local now = tonumber(clock[1]) + tonumber(clock[2]) / 1000000; " + "local start = tonumber(redis.call('GET', KEYS[1]) or now); " + "if now - start >= tonumber(ARGV[1]) then start = now end; " + "local ttl = math.max(1, math.ceil(tonumber(ARGV[1]) - (now - start))); " + "redis.call('SET', KEYS[1], start, 'EX', ttl); return ttl" + ) + ) + return math.ceil(_NUMBER.validate_python(await script(keys=(model.start_key,), args=(model.duration,)))) + local: Final = _LOCAL.validate_python(cache.in_memory_cache) + now: Final = time.time() + cached: Final = local.get_cache(model.start_key) + previous: Final = _NUMBER.validate_python(now if cached is None else cached) + start: Final = now if now - previous >= model.duration else previous + ttl: Final = max(1, math.ceil(model.duration - (now - start))) + local.delete_cache(model.start_key) + local.set_cache(model.start_key, start, ttl=ttl) + return ttl + + +@dataclass(slots=True) +class EvaluationAttempt: + cache: DualCache + model: _ModelBudget | None + total: dict[str, object] | None = None # mutable-ok: existing spend writer owns this reservation format + member: str = "" + input_cost: float = 0.0 + known_cost: float = 0.0 + model_applied_cost: float | None = None + lock: asyncio.Lock = field(default_factory=asyncio.Lock) + + async def settle(self, cost: float, cancelled: bool = False) -> None: + await _complete(self._settle(max(cost, self.input_cost if cancelled else 0.0))) + + async def _settle(self, cost: float) -> None: + async with self.lock: + self.known_cost = max(cost, self.known_cost) + try: + await reconcile_budget_reservation( + {**self.total, "finalized": False, "externally_settled": False} if self.total is not None else None, + self.known_cost, + finalize=False, + ) + if self.total is not None: + self.total["externally_settled"] = True + finally: + if self.model is not None: + delta: Final = max(self.known_cost - (self.model_applied_cost or 0.0), 0.0) + ttl: Final = await _model_window(self.cache, self.model) if delta else 1 + await model_budget_spend( + self.cache, + self.model.spend_key, + operation="settle", + member=self.member, + adjustment=delta, + ttl=ttl, + ) + self.model_applied_cost = self.known_cost + + async def renew(self, request_task: asyncio.Task[object] | None) -> None: + deadline: Final = time.monotonic() + 2 * litellm.request_timeout + while self.model_applied_cost is None and self.model is not None: + await asyncio.sleep(_LEASE_SECONDS / 2) + async with self.lock: + if self.model_applied_cost is not None: + return + try: + if time.monotonic() >= deadline: + raise TimeoutError("Evaluation reservation exceeded request timeout") + await model_budget_spend(self.cache, self.model.spend_key, operation="renew", member=self.member) + except Exception: # noqa: BLE001 # an active call must stop when its budget hold cannot be renewed + verbose_proxy_logger.exception("Unable to renew evaluation budget reservation") + if request_task is not None and not request_task.done(): + request_task.cancel() + return + + +async def reserve_evaluation_budget( + owner: EvaluationBillingOwner, + request: Mapping[str, object], + call_type: str, +) -> EvaluationAttempt | None: + return await _complete(_reserve(owner, request, call_type, asyncio.current_task())) + + +async def _reserve( + owner: EvaluationBillingOwner, + request: Mapping[str, object], + call_type: str, + request_task: asyncio.Task[object] | None, +) -> EvaluationAttempt | None: + from litellm.proxy import proxy_server as proxy + + llm_router: Final = proxy.llm_router + metadata: Final = _MAPPING.validate_python(request.get("litellm_metadata") or request.get("metadata") or {}) + logical_model: Final = str(metadata.get("model_group") or request["model"]) + resolved: Final = resolve_model_budget(logical_model, owner.user_model_max_budget or {}) + model: Final = ( + _ModelBudget( + model_budget_spend_cache_key( + Litellm_EntityType.USER, owner.user_id, resolved.budget_model, resolved.budget_config.budget_duration + ), + model_budget_start_time_cache_key( + Litellm_EntityType.USER, owner.user_id, resolved.budget_model, resolved.budget_config.budget_duration + ), + duration_in_seconds(str(resolved.budget_config.budget_duration)), + resolved.budget_config.max_budget, + ) + if resolved is not None + and resolved.budget_config.max_budget is not None + and math.isfinite(resolved.budget_config.max_budget) + and resolved.budget_config.max_budget >= 0 + else None + ) + total_budget: Final = owner.max_budget is not None and math.isfinite(owner.max_budget) + if not total_budget and model is None: + return None + if total_budget: + current: Final = await proxy.get_current_spend( + counter_key=f"spend:user:{owner.user_id}", fallback_spend=owner.spend, max_budget=owner.max_budget + ) + if current >= _NUMBER.validate_python(owner.max_budget): + raise litellm.BudgetExceededError( + current_cost=current, + max_budget=_NUMBER.validate_python(owner.max_budget), + entity_type=Litellm_EntityType.USER.value, + entity_id=owner.user_id, + ) + route: Final = next(route for route, types in API_ROUTE_TO_CALL_TYPES.items() if CallTypes(call_type) in types) + model_info: Final = _MAPPING.validate_python(request.get("model_info") or metadata.get("model_info") or {}) + deployment_id: Final = model_info.get("id") + deployment: Final = ( + llm_router.get_deployment(deployment_id) if llm_router is not None and isinstance(deployment_id, str) else None + ) + pricing_model: Final = ( + deployment.model_name + if deployment is not None + else ( + resolve_model_group_alias(llm_router.model_group_alias, logical_model) or logical_model + if llm_router is not None + else logical_model + ) + ) + body: Final = _REQUEST.validate_python( + {**request, "model": pricing_model, "metadata": {}, "litellm_metadata": {}, "tags": []} + ) + attempt: Final = EvaluationAttempt(cache=proxy.model_max_budget_limiter.dual_cache, model=model) + try: + attempt.total = await reserve_budget_for_request( + request_body=body, + route=route, + llm_router=llm_router, + valid_token=UserAPIKeyAuth(user_id=owner.user_id), + team_object=None, + user_object=LiteLLM_UserTable(user_id=owner.user_id, max_budget=owner.max_budget, spend=owner.spend), + prisma_client=proxy.prisma_client, + user_api_key_cache=proxy.user_api_key_cache, + proxy_logging_obj=proxy.proxy_logging_obj, + fail_closed_budget_enforcement=True, + request_task=request_task, + ) + estimate: Final = ( + _NUMBER.validate_python(attempt.total["reserved_cost"]) + if attempt.total is not None + else estimate_request_max_cost(body, route, llm_router) + ) + if estimate is None or not math.isfinite(estimate) or estimate < 0: + raise ValueError("Evaluation budget cannot be checked for an unpriced model") + attempt.member = f"{uuid.uuid4()}:{estimate}" + attempt.input_cost = ( + _NUMBER.validate_python(attempt.total["input_cost"]) + if attempt.total is not None + else estimate_request_input_cost(body, route, llm_router) or 0.0 + ) + if model is not None: + spend: Final = await model_budget_spend( + attempt.cache, model.spend_key, operation="reserve", member=attempt.member, limit=model.limit + ) + if spend > model.limit or spend - estimate >= model.limit: + raise litellm.BudgetExceededError( + current_cost=spend - estimate, + max_budget=model.limit, + entity_type=Litellm_EntityType.USER.value, + entity_id=owner.user_id, + ) + lease: Final = asyncio.create_task(attempt.renew(request_task)) + _LEASES.add(lease) + lease.add_done_callback(_LEASES.discard) + except Exception: + await attempt.settle(0.0) + raise + return attempt diff --git a/litellm/router_strategy/budget_limiter.py b/litellm/router_strategy/budget_limiter.py index 631b0c3df3d..f31b26d62d0 100644 --- a/litellm/router_strategy/budget_limiter.py +++ b/litellm/router_strategy/budget_limiter.py @@ -21,6 +21,7 @@ anthropic: import asyncio import builtins import logging +import math from collections.abc import Mapping, Sequence from datetime import datetime, timedelta, timezone from itertools import groupby @@ -590,7 +591,7 @@ class RouterBudgetLimiting(CustomLogger): else: # Within existing window - increment spend remaining_time: Final = ttl_seconds - (current_time - budget_start) - ttl_for_increment: Final = int(remaining_time) + ttl_for_increment: Final = max(1, math.ceil(remaining_time)) await self._increment_spend_in_current_window( spend_key=spend_key, response_cost=response_cost, ttl=ttl_for_increment diff --git a/litellm/utils.py b/litellm/utils.py index d72588e2b00..e91f6d64d02 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -45,7 +45,7 @@ from httpx import Proxy from httpx._utils import get_environment_proxies from openai.lib import _parsing, _pydantic from openai.types.chat.completion_create_params import ResponseFormat -from pydantic import BaseModel +from pydantic import BaseModel, TypeAdapter import litellm import litellm.litellm_core_utils @@ -1379,11 +1379,14 @@ def _schedule_async_success_logging( first-wins rule: the innermost wrapper's provider-shaped result is the one the spend log reads usage from, and a later wrapper never swaps in its client-shaped translation. """ + from litellm.litellm_core_utils.litellm_logging import evaluation_logging_snapshot + + receipt_logger: Final = evaluation_logging_snapshot(logging_obj) def _enqueue_async_logging() -> None: asyncio.create_task( _client_async_logging_helper( - logging_obj=logging_obj, + logging_obj=receipt_logger, result=result, start_time=start_time, end_time=end_time, @@ -1965,11 +1968,26 @@ def client(original_function): @wraps(original_function) async def wrapper_async(*args, **kwargs): + from litellm.litellm_core_utils.internal_call_metadata import ( + EVALUATION_BUDGET_RESERVATION_KEY, + EvaluationBillingOwner, + get_evaluation_billing_owner, + ) + from litellm.litellm_core_utils.litellm_logging import EvaluationBudgetInvocation, Logging + from litellm.llms.anthropic.pass_through.messages.handler import anthropic_messages_uses_bridge + print_args_passed_to_litellm(original_function, args, kwargs) start_time: Final = datetime.datetime.now() result = None _update_response_metadata: Final[_ResponseMetadataUpdater] = litellm_utils.update_response_metadata logging_obj: LiteLLMLoggingObject | None = kwargs.get("litellm_logging_obj", None) + evaluation_invocation: Final = ( + EvaluationBudgetInvocation() + if get_evaluation_billing_owner() is not None + or isinstance(logging_obj, Logging) + and logging_obj.evaluation_billing_owner is not None + else None + ) LLMCachingHandler: Final = _get_cached_llm_caching_handler() _llm_caching_handler: Final[LLMCachingHandler] = LLMCachingHandler( original_function=original_function, @@ -1982,7 +2000,21 @@ def client(original_function): kwargs["litellm_call_id"] = str(uuid.uuid4()) model: Final[str | None] = args[0] if len(args) > 0 else kwargs.get("model", None) - is_completion_with_fallbacks: Final = kwargs.get("fallbacks") is not None + is_evaluation_bridge: Final = ( + evaluation_invocation is not None + and TypeAdapter(str).validate_python(call_type) == CallTypes.anthropic_messages.value + and anthropic_messages_uses_bridge( + TypeAdapter(str).validate_python(model or ""), TypeAdapter(Mapping[str, object]).validate_python(kwargs) + ) + ) + is_completion_with_fallbacks: Final = call_type in ( + CallTypes.acompletion.value, + CallTypes.atext_completion.value, + ) and bool( + TypeAdapter(object).validate_python( + kwargs.get("fallbacks") if kwargs.get("fallbacks") is not None else litellm.model_fallbacks + ) + ) kwargs.pop("_is_litellm_internal_call", None) # discard if injected _is_litellm_internal_call: Final = is_internal_call.get() _deployment_call_end_time: datetime.datetime | None = None @@ -1993,6 +2025,36 @@ def client(original_function): # Type assertion: logging_obj is guaranteed to be non-None after function_setup assert logging_obj is not None, "logging_obj should not be None after function_setup" + if ( + evaluation_invocation is not None + and isinstance(logging_obj, Logging) + and isinstance(logging_obj.evaluation_billing_owner, EvaluationBillingOwner) + and not _is_litellm_internal_call + and logging_obj.evaluation_budget_invocation is None + ): + if not is_completion_with_fallbacks and not is_evaluation_bridge: + logging_obj.evaluation_budget_invocation = evaluation_invocation + logging_obj.evaluation_budget_reservation = None + prior_receipt: Final = TypeAdapter(Mapping[str, object]).validate_python(logging_obj.model_call_details) + logging_obj.model_call_details = { + key: value + for key, value in prior_receipt.items() + if not key.startswith("has_logged_") + and key not in ("response_cost", "standard_logging_object", "combined_usage_object", "cache_hit") + } + logging_obj.model_call_details[EVALUATION_BUDGET_RESERVATION_KEY] = None + if ( + evaluation_invocation is not None + and not _is_litellm_internal_call + and _is_streaming_request( + TypeAdapter(dict[str, object]).validate_python(kwargs), TypeAdapter(str).validate_python(call_type) + ) + ): + raise litellm.BadRequestError( + message="Shadow evaluation requires a non-streaming response", + model=TypeAdapter(str).validate_python(model or ""), + llm_provider="", + ) if not _is_litellm_internal_call: bind_budget_reservation_to_callbacks(logging_obj.litellm_params) @@ -2089,6 +2151,30 @@ def client(original_function): and _caching_handler_response.embedding_uncached_input is not None else kwargs ) + streaming_requested: Final = _is_streaming_request(kwargs=kwargs, call_type=call_type) + if ( + evaluation_invocation is not None + and isinstance(logging_obj, Logging) + and isinstance(logging_obj.evaluation_billing_owner, EvaluationBillingOwner) + and logging_obj.evaluation_budget_invocation is evaluation_invocation + ): + if streaming_requested: + raise litellm.BadRequestError( + message="Shadow evaluation requires a non-streaming response", + model=model or "", + llm_provider="", + ) + from litellm.proxy.spend_tracking.evaluation_budget import reserve_evaluation_budget + + evaluation_invocation.reservation = await reserve_evaluation_budget( + logging_obj.evaluation_billing_owner, + TypeAdapter(dict[str, object]).validate_python( + {**call_kwargs, "model": model, "messages": logging_obj.messages} + ), + TypeAdapter(str).validate_python(call_type), + ) + logging_obj.evaluation_budget_reservation = evaluation_invocation.reservation + logging_obj.model_call_details[EVALUATION_BUDGET_RESERVATION_KEY] = evaluation_invocation.reservation try: result = await original_function(*args, **call_kwargs) except Exception as deployment_error: @@ -2104,7 +2190,6 @@ def client(original_function): raise end_time = datetime.datetime.now() - streaming_requested: Final = _is_streaming_request(kwargs=kwargs, call_type=call_type) if streaming_requested or _is_converted_stream_result(result): logging_obj.stream = True logging_obj.model_call_details["stream"] = True @@ -2168,7 +2253,7 @@ def client(original_function): result=result, start_time=start_time, end_time=end_time, - is_completion_with_fallbacks=is_completion_with_fallbacks, + is_completion_with_fallbacks=is_completion_with_fallbacks or is_evaluation_bridge, is_litellm_internal_call=_is_litellm_internal_call, ) return _llm_caching_handler._combine_cached_embedding_response_with_api_result( @@ -2191,12 +2276,24 @@ def client(original_function): result=result, start_time=start_time, end_time=end_time, - is_completion_with_fallbacks=is_completion_with_fallbacks, + is_completion_with_fallbacks=is_completion_with_fallbacks or is_evaluation_bridge, is_litellm_internal_call=_is_litellm_internal_call, ) return result - except Exception as e: + except (Exception, asyncio.CancelledError) as e: + if ( + evaluation_invocation is not None + and isinstance(logging_obj, Logging) + and logging_obj.evaluation_budget_invocation is evaluation_invocation + and evaluation_invocation.reservation is not None + ): + await evaluation_invocation.reservation.settle( + logging_obj.recover_failure_cost(TypeAdapter(object).validate_python(result)), + cancelled=isinstance(e, asyncio.CancelledError), + ) + if isinstance(e, asyncio.CancelledError): + raise traceback_exception: Final = traceback.format_exc() # Reuse the timestamp taken right when the deployment call itself failed, before # the failure hook ran, so a slow callback doesn't inflate the reported duration. @@ -2216,7 +2313,10 @@ def client(original_function): call_type = original_function.__name__ num_retries, kwargs = _get_wrapper_num_retries(kwargs=kwargs, exception=e) - if call_type == CallTypes.acompletion.value: + sdk_retries_enabled: Final = not isinstance(logging_obj, Logging) or not isinstance( + logging_obj.evaluation_billing_owner, EvaluationBillingOwner + ) + if call_type == CallTypes.acompletion.value and sdk_retries_enabled: context_window_fallback_dict: Final = kwargs.get("context_window_fallback_dict", {}) _is_litellm_router_call = "model_group" in ( @@ -2251,7 +2351,7 @@ def client(original_function): kwargs["model"] = context_window_fallback_dict[model] result = await original_function(*args, **kwargs) return result - elif call_type == CallTypes.aresponses.value: + elif call_type == CallTypes.aresponses.value and sdk_retries_enabled: _is_litellm_router_call = "model_group" in ( kwargs.get("metadata") or {} ) # check if call from litellm.router/proxy @@ -2282,6 +2382,12 @@ def client(original_function): raise e finally: + if ( + evaluation_invocation is not None + and isinstance(logging_obj, Logging) + and logging_obj.evaluation_budget_invocation is evaluation_invocation + ): + logging_obj.evaluation_budget_invocation = None # Restore trace_id/session_id contextvars to their pre-call value once # this call (in this asyncio Task) is fully done - see # request_correlation_in_logs. Unlike wrapper()'s sync path, it's safe to diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index 2578cb7d78a..5f04c21966b 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -45,6 +45,7 @@ import litellm.router as litellm_router_module # noqa: E402 # same import-time import litellm.utils as litellm_utils_module # noqa: E402 # same import-time dependency from litellm._logging import ALL_LOGGERS # noqa: E402 # same import-time dependency from litellm.anthropic_beta_headers_manager import reload_beta_headers_config # noqa: E402 # same import-time dependency +from litellm.caching.caching import DualCache # noqa: E402 # same import-time dependency from litellm.litellm_core_utils.prompt_templates import factory as prompt_factory_module # noqa: E402 # same import-time dependency from litellm.litellm_core_utils.prompt_templates import ( # noqa: E402 # same import-time dependency image_handling as image_handling_module, @@ -332,3 +333,20 @@ def pytest_sessionfinish() -> None: _close_handler_if_needed(getattr(litellm, name, None)) _run_coroutine_if_needed(close_litellm_async_clients()) enable_socket() + + +@pytest.fixture +def evaluation_spend_cache(monkeypatch: pytest.MonkeyPatch) -> DualCache: + from litellm.proxy import proxy_server + from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache + from litellm.proxy.hooks.model_max_budget_limiter import _PROXY_VirtualKeyModelMaxBudgetLimiter + + cache: Final = DualCache() + monkeypatch.setattr(proxy_server, "spend_counter_cache", cache) + monkeypatch.setattr(proxy_server, "user_api_key_cache", UserApiKeyCache()) + monkeypatch.setattr(proxy_server, "prisma_client", None) + monkeypatch.setattr(proxy_server, "general_settings", {}) + monkeypatch.setattr(proxy_server, "llm_router", None) + monkeypatch.setattr(proxy_server, "model_max_budget_limiter", _PROXY_VirtualKeyModelMaxBudgetLimiter(cache)) + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + return cache diff --git a/tests/unit/integrations/test_lago.py b/tests/unit/integrations/test_lago.py new file mode 100644 index 00000000000..07485de2a60 --- /dev/null +++ b/tests/unit/integrations/test_lago.py @@ -0,0 +1,43 @@ +import json +from datetime import datetime +from typing import Final + +import pytest +import respx + +from litellm.integrations.lago import LagoLogger +from litellm.litellm_core_utils.internal_call_metadata import ( + EVALUATION_BILLING_OWNER_KEY, + EvaluationBillingOwner, + project_evaluation_billing_kwargs, +) + + +@pytest.mark.parametrize("charge_by", ("user_id", "team_id", "end_user_id")) +@pytest.mark.parametrize("evaluation", (False, True)) +def test_evaluation_receipts_charge_the_creator_regardless_of_lago_charge_by( + monkeypatch: pytest.MonkeyPatch, charge_by: str, evaluation: bool +) -> None: + for name, value in { + "LAGO_API_KEY": "test", + "LAGO_API_BASE": "https://lago.invalid", + "LAGO_API_EVENT_CODE": "usage", + "LAGO_API_CHARGE_BY": charge_by, + }.items(): + monkeypatch.setenv(name, value) + receipt: Final = project_evaluation_billing_kwargs( + { + EVALUATION_BILLING_OWNER_KEY: EvaluationBillingOwner("creator") if evaluation else None, + "response_cost": 0.25, + "litellm_params": { + "metadata": {"user_api_key_user_id": "user_id", "user_api_key_team_id": "team_id"}, + "proxy_server_request": {"body": {"user": "end_user_id"}}, + }, + } + ) + with respx.mock(assert_all_called=True) as transport: + endpoint: Final = transport.post("https://lago.invalid/api/v1/events").respond(200) + LagoLogger().log_success_event(receipt, {}, datetime(2026, 1, 1), datetime(2026, 1, 1)) + sent: Final = json.loads(endpoint.calls.last.request.content)["event"] + assert sent["external_subscription_id"] == ("creator" if evaluation else charge_by) + assert sent["properties"]["response_cost"] == 0.25 diff --git a/tests/unit/integrations/test_shadow_eval_logger.py b/tests/unit/integrations/test_shadow_eval_logger.py index e3f059a7941..ca92d604656 100644 --- a/tests/unit/integrations/test_shadow_eval_logger.py +++ b/tests/unit/integrations/test_shadow_eval_logger.py @@ -11,7 +11,7 @@ import pytest from pydantic import ValidationError from litellm.caching.in_memory_cache import InMemoryCache -from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY +from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY, LITELLM_PROXY_ADMIN_NAME from litellm.integrations.shadow_eval_logger import ( _MAX_CONCURRENT_SHADOW_TASKS, _MAX_ERROR_CHARS, @@ -27,6 +27,7 @@ from litellm.integrations.shadow_eval_logger import ( _unmask_preference, request_guardrail_fingerprint, ) +from litellm.proxy._types import LiteLLM_UserTable from litellm.types.guardrails import GuardrailEventHooks from litellm.types.utils import ( SHADOW_EVAL_JUDGE_CALL_ORIGIN, @@ -52,6 +53,7 @@ def _job(**overrides) -> ActiveShadowEvalJob: router_name="my-router", shadow_percentage=100.0, judge_model="judge-model", + created_by="evaluation-admin", max_turns=200, ends_at=datetime.now(timezone.utc) + timedelta(days=1), attempts=0, @@ -74,6 +76,7 @@ def _prisma(jobs=(), attempt_counts=(), attempt_costs=()) -> MagicMock: ] ) prisma.db.litellm_shadowevalattempt.create = AsyncMock() + prisma.db.litellm_usertable.find_unique = AsyncMock(return_value=LiteLLM_UserTable(user_id="evaluation-admin")) return prisma @@ -90,6 +93,7 @@ def _job_record(job: ActiveShadowEvalJob, target_type="key", target_id="key-hash baseline_model=job.baseline_model, shadow_percentage=job.shadow_percentage, judge_model=job.judge_model, + created_by=job.created_by, max_turns=job.max_turns, max_budget=job.max_budget, ends_at=job.ends_at, @@ -402,7 +406,7 @@ class TestSurfaceNormalization: assert shadow_call["tools"][0]["type"] == "function" assert shadow_call["tools"][0]["function"]["name"] == "get_weather" assert "stop_sequences" not in shadow_call - assert "stream" not in shadow_call + assert shadow_call["stream"] is False assert shadow_call["metadata"][INTERNAL_CALL_ORIGIN_METADATA_KEY] == SHADOW_EVAL_ROUTER_CALL_ORIGIN async def test_responses_arm_translates_wire_body_params_and_drops_surface_only_keys(self): @@ -1029,7 +1033,7 @@ class TestSuccessHookSkipChain: shadow_call = router.acompletion.call_args_list[0].kwargs assert shadow_call["temperature"] == 0.5 - assert "stream" not in shadow_call + assert shadow_call["stream"] is False create = prisma.db.litellm_shadowevalattempt.create create.assert_awaited_once() row = create.call_args.kwargs["data"] @@ -1993,6 +1997,7 @@ class TestShadowPipeline: shadow_call = router.acompletion.call_args_list[0].kwargs judge_call = router.acompletion.call_args_list[1].kwargs for call in (shadow_call, judge_call): + assert call["stream"] is False assert call["num_retries"] == 0 assert call["fallbacks"] == [] assert call["metadata"]["user_api_key_hash"] == "key-hash" @@ -2521,3 +2526,29 @@ class TestSamplingFunnel: assert logger._test_funnel == [] prisma.db.litellm_shadowevalattempt.create.assert_not_awaited() + + +@pytest.mark.asyncio +@pytest.mark.usefixtures("evaluation_spend_cache") +@pytest.mark.parametrize( + ("created_by", "failure"), + ( + ("missing-creator", None), + ("unreadable-creator", RuntimeError("creator unavailable")), + (LITELLM_PROXY_ADMIN_NAME, None), + (None, None), + ), +) +async def test_unavailable_evaluation_creator_withholds_without_calling_a_provider( + created_by: str | None, failure: Exception | None +) -> None: + prisma: Final = _prisma() + prisma.db.litellm_usertable.find_unique = AsyncMock(return_value=None, side_effect=failure) + router: Final = _router() + job: Final = _job(created_by=created_by) + logger: Final = _logger(router=router, prisma=prisma, jobs=(job,)) + await logger.async_log_success_event(_success_kwargs(), RESPONSE, None, None) + await _drain(logger) + assert router.acompletion.call_count == 0 + assert logger._test_funnel == [("job-1", "withheld")] + prisma.db.litellm_shadowevalattempt.create.assert_not_awaited() diff --git a/tests/unit/litellm_core_utils/test_litellm_logging.py b/tests/unit/litellm_core_utils/test_litellm_logging.py index 4b25da2ff79..f16b94f088e 100644 --- a/tests/unit/litellm_core_utils/test_litellm_logging.py +++ b/tests/unit/litellm_core_utils/test_litellm_logging.py @@ -9165,3 +9165,228 @@ def test_signoz_dispatch_requires_an_endpoint(monkeypatch): logging_module._in_memory_loggers.clear() monkeypatch.delenv("LITELLM_OTEL_V2", raising=False) is_otel_v2_enabled.cache_clear() + + +@pytest.mark.asyncio +@pytest.mark.usefixtures("evaluation_spend_cache") +@pytest.mark.parametrize("dispatch", ("cache", "success", "failure")) +async def test_queued_evaluation_callbacks_keep_their_receipt_after_logger_reuse( + monkeypatch: pytest.MonkeyPatch, dispatch: Literal["cache", "success", "failure"] +) -> None: + from concurrent.futures import ThreadPoolExecutor + from queue import Queue + from threading import Event + + from pydantic import TypeAdapter + + from litellm.integrations.s3 import S3Logger + from litellm.litellm_core_utils import litellm_logging, thread_pool_executor + from litellm.litellm_core_utils.internal_call_metadata import EvaluationBillingOwner, evaluation_billing_context + from litellm.types.utils import Usage + + receipts: Final = Queue[Mapping[str, object]]() + payload: Final = TypeAdapter(Mapping[str, object]) + + def upload(**request: object) -> None: + body: Final = request["Body"] + assert isinstance(body, str) + receipts.put_nowait(payload.validate_json(body)) + + def failure(kwargs: Mapping[str, object], response: object, start: object, end: object) -> None: + receipts.put_nowait(payload.validate_python(kwargs["standard_logging_object"])) + + with patch("boto3.client", return_value=MagicMock(put_object=upload)): + sink: Final = S3Logger(s3_bucket_name="evaluation-receipts") + monkeypatch.setattr(litellm_logging, "s3Logger", sink) + monkeypatch.setattr(litellm, "success_callback", ["s3"]) + monkeypatch.setattr(litellm, "failure_callback", [failure]) + now: Final = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) + result: Final = ModelResponse( + model="openai/evaluation-receipt-model", usage=Usage(prompt_tokens=1, completion_tokens=1, total_tokens=2) + ) + with evaluation_billing_context(EvaluationBillingOwner("creator")): + logger: Final = LitellmLogging( + model="openai/evaluation-receipt-model", + messages=[{"role": "user", "content": "hello"}], + stream=False, + call_type="acompletion", + start_time=now, + litellm_call_id="first-call", + function_id="first-call", + ) + logger.update_environment_variables( + litellm_params={"acompletion": True, "metadata": {"model_group": "first-group"}}, optional_params={} + ) + logger.record_partial_usage_for_failure(Usage(prompt_tokens=1, completion_tokens=1, total_tokens=2), 0.3) + release: Final = Event() + with ThreadPoolExecutor(max_workers=1) as worker: + worker.submit(release.wait) + monkeypatch.setattr(litellm_logging, "executor", worker) + monkeypatch.setattr(thread_pool_executor, "executor", worker) + try: + if dispatch == "cache": + logger.handle_sync_success_callbacks_for_async_calls(result, now, now, cache_hit=True) + elif dispatch == "success": + await logger.dispatch_success_handlers(result, start_time=now, end_time=now, cache_hit=False) + else: + await logger.dispatch_failure_handlers(RuntimeError("first failure"), "first failure") + later_params: Final = {"acompletion": True, "metadata": {"model_group": "later-group"}} + later: Final = { + **logger.model_call_details, + "model": "later-model", + "response_cost": 9.0, + "cache_hit": False, + "exception": RuntimeError("later failure"), + "litellm_params": later_params, + } + logger.model_call_details = later + logger.model = "later-model" + logger.litellm_params = later_params + untouched: Final = later.copy() + finally: + release.set() + assert receipts.qsize() == 1 + receipt: Final = receipts.get_nowait() + assert receipt["model_group"] == "first-group" + assert receipt["response_cost"] == pytest.approx(0.0 if dispatch == "cache" else 0.3) + assert receipt["cache_hit"] is (dispatch == "cache") + assert payload.validate_python(receipt["metadata"])["user_api_key_user_id"] == "creator" + assert receipt["error_str"] == ("first failure" if dispatch == "failure" else None) + assert logger.model_call_details == untouched + + +@pytest.mark.asyncio +@pytest.mark.usefixtures("evaluation_spend_cache") +@pytest.mark.parametrize("api", ("chat", "messages")) +async def test_deferred_evaluation_receipts_preserve_routing_metadata_on_real_router_reuse( + monkeypatch: pytest.MonkeyPatch, api: Literal["chat", "messages"] +) -> None: + from concurrent.futures import ThreadPoolExecutor + from queue import Queue + from threading import Event + + import respx + from pydantic import TypeAdapter + + from litellm.integrations.s3 import S3Logger + from litellm.litellm_core_utils import litellm_logging, thread_pool_executor + from litellm.litellm_core_utils.internal_call_metadata import EvaluationBillingOwner, evaluation_billing_context + from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER + from litellm.proxy import proxy_server + from litellm.types.utils import Usage + + sync_receipts: Final = Queue[Mapping[str, object]]() + async_receipts: Final = asyncio.Queue[Mapping[str, object]]() + payload: Final = TypeAdapter(Mapping[str, object]) + + def upload(**request: object) -> None: + body: Final = request["Body"] + assert isinstance(body, str) + sync_receipts.put_nowait(payload.validate_json(body)) + + async def capture(kwargs: Mapping[str, object], response: object, start: object, end: object) -> None: + async_receipts.put_nowait(payload.validate_python(kwargs["standard_logging_object"])) + + with patch("boto3.client", return_value=MagicMock(put_object=upload)): + sink: Final = S3Logger(s3_bucket_name="evaluation-receipts") + monkeypatch.setattr(litellm_logging, "s3Logger", sink) + monkeypatch.setattr(litellm, "success_callback", ["s3"]) + model: Final = "hosted_vllm/evaluation-receipt" + monkeypatch.setitem( + litellm.model_cost, + model, + { + "input_cost_per_token": 0.001, + "output_cost_per_token": 0.002, + "max_input_tokens": 1000, + "max_output_tokens": 1000, + "litellm_provider": "hosted_vllm", + "mode": "chat", + }, + ) + router: Final = litellm.Router( + model_list=[ + { + "model_name": group, + "litellm_params": { + "model": model, + "api_key": "test", + "api_base": "https://evaluation.invalid/v1", + "input_cost_per_token": input_cost, + "output_cost_per_token": output_cost, + }, + "model_info": {"id": group + "-id"}, + } + for group, input_cost, output_cost in (("first-group", 0.001, 0.002), ("later-group", 0.003, 0.004)) + ], + num_retries=0, + ) + monkeypatch.setattr(proxy_server, "llm_router", router) + create: Final = router.acompletion if api == "chat" else router.aanthropic_messages + metadata: Final = {"user_api_key_user_id": "caller", "user_api_key_team_id": "caller-team"} + release: Final = Event() + with ( + respx.mock(assert_all_called=True) as transport, + evaluation_billing_context(EvaluationBillingOwner("creator", max_budget=1)), + ThreadPoolExecutor(max_workers=1) as worker, + ): + worker.submit(release.wait) + monkeypatch.setattr(litellm_logging, "executor", worker) + monkeypatch.setattr(thread_pool_executor, "executor", worker) + upstream: Final = transport.post("https://evaluation.invalid/v1/chat/completions").respond( + 200, + json=ModelResponse( + model="evaluation-receipt", + choices=[{"message": {"role": "assistant", "content": "ok"}}], + usage=Usage(prompt_tokens=10, completion_tokens=1, total_tokens=11), + ).model_dump(), + ) + logger: Final = LitellmLogging( + model=model, + messages=[], + stream=False, + call_type="acompletion" if api == "chat" else "anthropic_messages", + start_time=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc), + litellm_call_id="reused-evaluation", + function_id="test", + dynamic_async_success_callbacks=[capture], + ) + logger._defer_async_logging = True + request: Final = { + "messages": [{"role": "user", "content": "hello"}], + "max_tokens": 10, + "metadata" if api == "chat" else "litellm_metadata": metadata, + "litellm_logging_obj": logger, + "fallbacks": [], + } + try: + await create(model="first-group", **request) + first: Final = logger._enqueue_deferred_logging + assert first is not None + logger._enqueue_deferred_logging = None + await create(model="later-group", **request) + second: Final = logger._enqueue_deferred_logging + assert second is not None + first() + second() + received: Final = ( + await asyncio.wait_for(async_receipts.get(), 10), + await asyncio.wait_for(async_receipts.get(), 10), + ) + await GLOBAL_LOGGING_WORKER.flush() + finally: + release.set() + assert upstream.call_count == 2 and async_receipts.empty() and sync_receipts.qsize() == 2 + for receipts in (received, (sync_receipts.get_nowait(), sync_receipts.get_nowait())): + assert {receipt["model_group"]: receipt["model_id"] for receipt in receipts} == { + "first-group": "first-group-id", + "later-group": "later-group-id", + } + assert {receipt["model_group"]: receipt["response_cost"] for receipt in receipts} == pytest.approx( + {"first-group": 10 * 0.001 + 0.002, "later-group": 10 * 0.003 + 0.004} + ) + for receipt in receipts: + assert payload.validate_python(receipt["metadata"])["user_api_key_user_id"] == "creator" + assert payload.validate_python(receipt["metadata"])["user_api_key_team_id"] is None + assert metadata["model_group"] == "later-group" + assert metadata["user_api_key_user_id"] == "caller" and metadata["user_api_key_team_id"] == "caller-team" diff --git a/tests/unit/litellm_core_utils/test_llm_judge.py b/tests/unit/litellm_core_utils/test_llm_judge.py index 3bcfde76450..9780b8aaeca 100644 --- a/tests/unit/litellm_core_utils/test_llm_judge.py +++ b/tests/unit/litellm_core_utils/test_llm_judge.py @@ -1,12 +1,15 @@ """Unit tests for the shared LLM-judge primitives: verdict parsing, router resolution, dispatch.""" import json -from unittest.mock import AsyncMock, MagicMock +from typing import Final +from unittest.mock import AsyncMock import pytest +import respx import litellm from litellm.litellm_core_utils.llm_judge import ( + default_router_provider, extract_text_from_content, judge_acompletion, judge_target, @@ -106,6 +109,30 @@ async def test_judge_acompletion_falls_back_to_sdk_for_unconfigured_model(monkey assert sdk.call_args.kwargs["drop_params"] is True +@pytest.mark.asyncio +@pytest.mark.usefixtures("evaluation_spend_cache") +async def test_sdk_judge_failure_does_not_inherit_a_global_fallback(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(litellm, "model_fallbacks", ["openai/judge-fallback"]) + with respx.mock(assert_all_called=False) as transport: + primary: Final = transport.post("https://judge.test/v1/chat/completions", json__model="judge-primary").respond( + 400, json={"error": {"message": "judge rejected", "type": "invalid_request_error"}} + ) + fallback: Final = transport.post( + "https://judge.test/v1/chat/completions", json__model="judge-fallback" + ).respond(200, json=litellm.ModelResponse().model_dump()) + with pytest.raises(litellm.BadRequestError, match="judge rejected"): + await judge_acompletion( + default_router_provider(), + "openai/judge-primary", + [{"role": "user", "content": "judge this"}], + api_key="test", + api_base="https://judge.test/v1", + max_tokens=10, + ) + assert primary.call_count == 1 + assert not fallback.called + + @pytest.mark.parametrize( "model,expected", [ diff --git a/tests/unit/proxy/spend_tracking/test_evaluation_budget.py b/tests/unit/proxy/spend_tracking/test_evaluation_budget.py new file mode 100644 index 00000000000..ee1bc29dccb --- /dev/null +++ b/tests/unit/proxy/spend_tracking/test_evaluation_budget.py @@ -0,0 +1,242 @@ +import asyncio +import time +from collections.abc import Mapping +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Final +from unittest.mock import patch + +import pytest +import respx + +import litellm +from litellm.caching.caching import DualCache +from litellm.caching.in_memory_cache import InMemoryCache +from litellm.litellm_core_utils.internal_call_metadata import EvaluationBillingOwner, evaluation_billing_context +from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER +from litellm.proxy import proxy_server +from litellm.proxy.spend_tracking.budget_reservation import estimate_request_max_cost +from litellm.proxy.spend_tracking.evaluation_budget import ( + EvaluationAttempt, + _complete, + model_budget_spend, + reserve_evaluation_budget, +) +from litellm.types.utils import ModelResponse, Usage + + +@pytest.mark.asyncio +@pytest.mark.parametrize("scope", ("total", "model", "both")) +async def test_concurrent_evaluations_settle_once_without_spending_another_attempt( + evaluation_spend_cache: DualCache, monkeypatch: pytest.MonkeyPatch, scope: str +) -> None: + monkeypatch.setattr(evaluation_spend_cache, "in_memory_cache", InMemoryCache(clock=lambda: 0.0)) + model: Final = "openai/evaluation-budget-test" + monkeypatch.setitem(litellm.model_cost, model, {"input_cost_per_token": 0.001, "output_cost_per_token": 0.002}) + request: Final = {"model": model, "messages": [{"role": "user", "content": "hello"}], "max_tokens": 10} + estimate: Final = estimate_request_max_cost(request, "/chat/completions", None) + assert estimate is not None and estimate > 0 + owner: Final = EvaluationBillingOwner( + "creator", + {model: {"max_budget": estimate * 1.5, "budget_duration": "1d"}} if scope != "total" else None, + max_budget=estimate * 1.5 if scope != "model" else None, + ) + attempts: Final = await asyncio.gather( + *(reserve_evaluation_budget(owner, request, "acompletion") for _ in range(2)), return_exceptions=True + ) + assert sum(isinstance(result, litellm.BudgetExceededError) for result in attempts) == 1 + admitted: Final = next(result for result in attempts if not isinstance(result, BaseException)) + assert admitted is not None + await admitted.settle(estimate / 4) + later: Final = await reserve_evaluation_budget(owner, request, "acompletion") + assert later is not None + for cost in (estimate / 4, 0, estimate / 2): + await admitted.settle(cost) + await proxy_server.increment_spend_counters( + token=None, team_id=None, user_id="creator", response_cost=cost, budget_reservation=admitted.total + ) + await later.settle(0) + await proxy_server.model_max_budget_limiter.async_log_success_event( + { + "litellm_params": {"metadata": {"user_api_key_user_model_max_budget": owner.user_model_max_budget}}, + "standard_logging_object": { + "model": model, + "response_cost": 0.01, + "metadata": {"user_api_key_user_id": owner.user_id}, + }, + }, + None, + None, + None, + ) + keys: Final = {"total": "spend:user:creator", "model": f"user_model_spend:creator:{model}:1d"} + for kind in keys if scope == "both" else (scope,): + assert await evaluation_spend_cache.async_get_cache(keys[kind]) == pytest.approx( + estimate / 2 + (0.01 if kind == "model" else 0) + ) + + +@pytest.mark.asyncio +async def test_auto_router_prices_its_selected_deployment_and_bills_the_creator( + evaluation_spend_cache: DualCache, monkeypatch: pytest.MonkeyPatch +) -> None: + rates: Final = {"input_cost_per_token": 0.001, "output_cost_per_token": 0.002} + router: Final = litellm.Router( + model_list=[ + { + "model_name": "leaf", + "litellm_params": {"model": "openai/private-evaluation", "api_key": "test", **rates}, + "model_info": {"id": "selected", "max_input_tokens": 1000, "max_output_tokens": 10}, + }, + { + "model_name": "router", + "litellm_params": { + "model": "auto_router/complexity_router", + "complexity_router_config": { + "classifier_type": "heuristic", + "tiers": dict.fromkeys(("SIMPLE", "MEDIUM", "COMPLEX", "REASONING"), "leaf"), + }, + }, + }, + ], + num_retries=0, + ) + monkeypatch.setattr(proxy_server, "llm_router", router) + owner: Final = EvaluationBillingOwner("creator", {"router": {"max_budget": 1, "budget_duration": "1d"}}, 1) + receipts: Final[asyncio.Queue[Mapping[str, object]]] = asyncio.Queue() + source: Final = {"user_api_key_user_id": "source", "user_api_key_team_id": "team", "agent_id": "agent"} + + async def capture(kwargs: Mapping[str, object], response: object, start: datetime, end: datetime) -> None: + receipts.put_nowait(kwargs) + + with respx.mock(assert_all_called=True) as transport: + transport.post("https://api.openai.com/v1/chat/completions").respond( + 200, + json=ModelResponse( + model="private-evaluation", + choices=[{"message": {"role": "assistant", "content": "ok"}}], + usage=Usage(prompt_tokens=10, completion_tokens=2, total_tokens=12), + ).model_dump(), + ) + + with evaluation_billing_context(owner): + await router.acompletion( + model="router", + messages=[{"role": "user", "content": "hello"}], + max_tokens=10, + metadata=source, + fallbacks=[], + success_callback=[capture], + ) + receipt: Final = await asyncio.wait_for(receipts.get(), 10) + await GLOBAL_LOGGING_WORKER.flush() + assert (receipt["user"], receipt["agent_id"], receipt["request_tags"]) == ("creator", None, []) + assert source["user_api_key_user_id"] == "source" and source["agent_id"] == "agent" + actual: Final = 10 * rates["input_cost_per_token"] + 2 * rates["output_cost_per_token"] + assert receipt["response_cost"] == pytest.approx(actual) + for key in ("spend:user:creator", "user_model_spend:creator:router:1d"): + assert await evaluation_spend_cache.async_get_cache(key) == pytest.approx(actual) + + +@dataclass(slots=True) +class _Clock: + seconds: float = 0.0 + + +@pytest.mark.asyncio +async def test_expired_model_hold_cannot_release_or_renew_another_requests_budget() -> None: + clock: Final = _Clock() + cache: Final = DualCache(in_memory_cache=InMemoryCache(clock=lambda: clock.seconds)) + await model_budget_spend(cache, "budget", operation="reserve", member="old:0.2") + clock.seconds = 30 + assert await model_budget_spend(cache, "budget", operation="reserve", member="new:0.3") == pytest.approx(0.5) + clock.seconds = 61 + assert await model_budget_spend(cache, "budget") == pytest.approx(0.3) + assert await model_budget_spend(cache, "budget", operation="settle", member="old:0.2") == pytest.approx(0.3) + with pytest.raises(RuntimeError, match="Evaluation reservation expired"): + await model_budget_spend(cache, "budget", operation="renew", member="old:0.2") + await model_budget_spend(cache, "budget", operation="renew", member="new:0.3") + clock.seconds = 92 + assert await model_budget_spend(cache, "budget") == pytest.approx(0.3) + assert await model_budget_spend(cache, "budget", operation="settle", member="new:0.3") == 0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("phase", ("admission", "settlement")) +async def test_repeated_cancellation_drains_budget_operations_before_returning( + evaluation_spend_cache: DualCache, monkeypatch: pytest.MonkeyPatch, phase: str +) -> None: + model: Final = "openai/cancelled-evaluation" + monkeypatch.setitem(litellm.model_cost, model, {"input_cost_per_token": 0.001, "output_cost_per_token": 0.002}) + request: Final = {"model": model, "messages": [{"role": "user", "content": "hello"}], "max_tokens": 10} + owner: Final = EvaluationBillingOwner("creator", {model: {"max_budget": 1, "budget_duration": "1d"}}, 1) + existing: Final = await reserve_evaluation_budget(owner, request, "acompletion") if phase == "settlement" else None + entered: Final = asyncio.Event() + proceed: Final = asyncio.Event() + + async def delayed_write() -> EvaluationAttempt | None: + attempt: Final = existing or await reserve_evaluation_budget(owner, request, "acompletion") + assert attempt is not None + entered.set() + await proceed.wait() + if existing is None: + return attempt + await attempt.settle(0.005) + return None + + pending: Final = asyncio.create_task(_complete(delayed_write())) + await asyncio.wait_for(entered.wait(), 5) + pending.cancel() + await asyncio.sleep(0) + pending.cancel() + assert not pending.done() + proceed.set() + with pytest.raises(asyncio.CancelledError): + await pending + expected: Final = 0.005 if phase == "settlement" else 0 + assert await evaluation_spend_cache.async_get_cache("spend:user:creator") == pytest.approx(expected) + assert await model_budget_spend(evaluation_spend_cache, f"user_model_spend:creator:{model}:1d") == pytest.approx( + expected + ) + + +@pytest.mark.asyncio +async def test_evaluation_settlement_preserves_spend_until_the_shared_budget_window_ends( + evaluation_spend_cache: DualCache, monkeypatch: pytest.MonkeyPatch +) -> None: + clock: Final = _Clock() + epoch: Final = 1000.0 + monkeypatch.setattr(time, "time", lambda: epoch + 0.2 + clock.seconds) + monkeypatch.setattr(evaluation_spend_cache, "in_memory_cache", InMemoryCache(clock=lambda: clock.seconds)) + model: Final = "openai/shared-window" + monkeypatch.setitem(litellm.model_cost, model, {"input_cost_per_token": 0.001, "output_cost_per_token": 0.002}) + owner: Final = EvaluationBillingOwner("creator", {model: {"max_budget": 1, "budget_duration": "10s"}}) + key: Final = f"user_model_spend:creator:{model}:10s" + evaluation_spend_cache.in_memory_cache.set_cache(key, 0.1, ttl=1) + evaluation_spend_cache.in_memory_cache.set_cache(f"user_model_budget_start_time:creator:{model}:10s", epoch, ttl=10) + attempt: Final = await reserve_evaluation_budget( + owner, {"model": model, "messages": [{"role": "user", "content": "hello"}], "max_tokens": 10}, "acompletion" + ) + assert attempt is not None + await attempt.settle(0.02) + with patch("litellm.router_strategy.budget_limiter.datetime") as wall_clock: + wall_clock.now.return_value = datetime.fromtimestamp(epoch + 0.2, timezone.utc) + await proxy_server.model_max_budget_limiter.async_log_success_event( + { + "litellm_params": {"metadata": {"user_api_key_user_model_max_budget": owner.user_model_max_budget}}, + "standard_logging_object": { + "model": model, + "response_cost": 0.01, + "metadata": {"user_api_key_user_id": owner.user_id}, + }, + }, + None, + None, + None, + ) + clock.seconds = 2 + assert await model_budget_spend(evaluation_spend_cache, key) == pytest.approx(0.13) + clock.seconds = 9.9 + assert await model_budget_spend(evaluation_spend_cache, key) == pytest.approx(0.13) + clock.seconds = 10.1 + assert await model_budget_spend(evaluation_spend_cache, key) == 0 diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py index a72aec07766..9a1fef32132 100644 --- a/tests/unit/test_utils.py +++ b/tests/unit/test_utils.py @@ -29,13 +29,16 @@ from litellm._logging import ( trace_id_var, verbose_logger, ) -from litellm.caching.caching import Cache +from litellm.caching.caching import Cache, DualCache from litellm.caching.caching_handler import _PENDING_CACHE_WRITES from litellm.caching.in_memory_cache import InMemoryCache from litellm.constants import DEFAULT_MOCK_RESPONSE_COMPLETION_TOKEN_COUNT from litellm.integrations.custom_guardrail import CustomGuardrail from litellm.integrations.custom_logger import CustomLogger from litellm.litellm_core_utils.get_litellm_params import get_litellm_params +from litellm.litellm_core_utils.internal_call_metadata import EvaluationBillingOwner, evaluation_billing_context +from litellm.litellm_core_utils.litellm_logging import Logging +from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER from litellm.litellm_core_utils.thread_pool_executor import executor as logging_executor from litellm.llms.base_llm.base_model_iterator import MockResponseIterator from litellm.proxy.utils import is_valid_api_key @@ -6558,3 +6561,342 @@ def test_function_setup_never_logs_the_ocr_data_uri_payload() -> None: assert logged == [{"role": "user", "content": f"data:application/pdf;base64 ({len(payload)} chars)"}] assert payload not in str(logged) + + +@pytest.fixture +def evaluation_model(monkeypatch: pytest.MonkeyPatch) -> str: + model: Final = "hosted_vllm/evaluation-test" + for deployment in (model, model + "-fallback", "anthropic/evaluation-native"): + monkeypatch.setitem( + litellm.model_cost, + deployment, + { + "input_cost_per_token": 0.001, + "output_cost_per_token": 0.002, + "max_input_tokens": 1000, + "max_output_tokens": 1000, + "litellm_provider": deployment.split("/", 1)[0], + "mode": "chat", + }, + ) + return model + + +@pytest.mark.asyncio +@pytest.mark.parametrize("api", ("chat", "messages", "messages_responses", "embedding")) +@pytest.mark.parametrize("scope", ("total", "model")) +async def test_evaluation_budget_blocks_paid_leaf_calls_before_transport( + evaluation_spend_cache: DualCache, evaluation_model: str, api: str, scope: str +) -> None: + model: Final = "openai/gpt-5.6-luna" if api == "messages_responses" else evaluation_model + owner: Final = EvaluationBillingOwner( + "creator", + {model: {"max_budget": 0, "budget_duration": "1d"}} if scope == "model" else None, + max_budget=0 if scope == "total" else None, + ) + create: Final = { + "chat": litellm.acompletion, + "messages": litellm.anthropic.messages.acreate, + "messages_responses": litellm.anthropic.messages.acreate, + "embedding": litellm.aembedding, + }[api] + body: Final = ( + {"input": ["hello"]} + if api == "embedding" + else {"messages": [{"role": "user", "content": "hello"}], "max_tokens": 10} + ) + with respx.mock(assert_all_called=False) as transport, evaluation_billing_context(owner): + upstream: Final = transport.post(url__startswith="https://evaluation.invalid/").respond(500) + with pytest.raises(litellm.BudgetExceededError): + await create( + model=model, + api_base="https://evaluation.invalid/v1", + api_key="test", + metadata={"internal_call_origin": "autorouter_classifier"}, + fallbacks=[], + num_retries=0, + **body, + ) + assert upstream.call_count == 0 + assert (await evaluation_spend_cache.async_get_cache("spend:user:creator") or 0) == 0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("global_fallbacks", (False, True)) +async def test_messages_fallback_checks_its_own_budget_before_a_second_provider_call( + evaluation_spend_cache: DualCache, + evaluation_model: str, + monkeypatch: pytest.MonkeyPatch, + global_fallbacks: bool, +) -> None: + fallback: Final = evaluation_model + "-fallback" + monkeypatch.setattr(litellm, "model_fallbacks", [fallback] if global_fallbacks else None) + owner: Final = EvaluationBillingOwner("creator", {fallback: {"max_budget": 0, "budget_duration": "1d"}}, 1) + with respx.mock(assert_all_called=True) as transport, evaluation_billing_context(owner): + upstream: Final = transport.post("https://evaluation.invalid/v1/chat/completions").respond( + 500, json={"error": {"message": "unavailable", "type": "server_error"}} + ) + with pytest.raises(Exception, match="Max budget: 0"): + await litellm.anthropic.messages.acreate( + model=evaluation_model, + messages=[{"role": "user", "content": "hi"}], + max_tokens=10, + fallbacks=None if global_fallbacks else [fallback], + api_base="https://evaluation.invalid/v1", + api_key="test", + num_retries=0, + max_retries=0, + ) + assert upstream.call_count == 1 + assert (await evaluation_spend_cache.async_get_cache("spend:user:creator") or 0) == 0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("global_fallbacks", (False, True)) +async def test_native_messages_keep_creator_admission_when_chat_fallbacks_are_configured( + evaluation_spend_cache: DualCache, + evaluation_model: str, + monkeypatch: pytest.MonkeyPatch, + global_fallbacks: bool, +) -> None: + model: Final = "anthropic/evaluation-native" + monkeypatch.setattr(litellm, "model_fallbacks", [evaluation_model] if global_fallbacks else None) + owner: Final = EvaluationBillingOwner("creator", {model: {"max_budget": 0, "budget_duration": "1d"}}, 1) + with respx.mock(assert_all_called=False) as transport, evaluation_billing_context(owner): + upstream: Final = transport.post(url__startswith="https://evaluation.invalid/").respond(500) + with pytest.raises(litellm.BudgetExceededError): + await litellm.anthropic.messages.acreate( + model=model, + messages=[{"role": "user", "content": "hello"}], + max_tokens=10, + api_base="https://evaluation.invalid/v1", + api_key="test", + num_retries=0, + fallbacks=None if global_fallbacks else [evaluation_model], + ) + assert upstream.call_count == 0 + assert (await evaluation_spend_cache.async_get_cache("spend:user:creator") or 0) == 0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("api", ("chat", "messages")) +async def test_delayed_evaluation_receipts_survive_logger_reuse_and_a_cache_hit( + evaluation_spend_cache: DualCache, evaluation_model: str, monkeypatch: pytest.MonkeyPatch, api: str +) -> None: + monkeypatch.setattr(litellm, "cache", Cache(type="local")) + owner: Final = EvaluationBillingOwner("creator", max_budget=1) + receipts: Final[asyncio.Queue[Mapping[str, object]]] = asyncio.Queue() + requests: Final[asyncio.Queue[httpx.Request]] = asyncio.Queue() + + async def capture(kwargs: Mapping[str, object], response: object, start: datetime, end: datetime) -> None: + receipts.put_nowait(kwargs) + + def upstream(request: httpx.Request) -> httpx.Response: + requests.put_nowait(request) + return httpx.Response( + 200, + json=ModelResponse( + model="evaluation-test", + choices=[{"message": {"role": "assistant", "content": "ok"}}], + usage=Usage(prompt_tokens=10, completion_tokens=requests.qsize(), total_tokens=10 + requests.qsize()), + ).model_dump(), + ) + + with respx.mock(assert_all_called=True) as transport, evaluation_billing_context(owner): + transport.post("https://evaluation.invalid/v1/chat/completions").mock(side_effect=upstream) + logger: Final = Logging( + model=evaluation_model, + messages=[], + stream=False, + call_type="acompletion", + start_time=datetime.now(), + litellm_call_id="evaluation", + function_id="test", + dynamic_async_success_callbacks=[capture], + ) + logger._defer_async_logging = True + create: Final = litellm.acompletion if api == "chat" else litellm.anthropic.messages.acreate + request: Final = { + "model": evaluation_model, + "messages": [{"role": "user", "content": "one"}], + "max_tokens": 10, + "api_base": "https://evaluation.invalid/v1", + "api_key": "test", + "num_retries": 0, + "litellm_logging_obj": logger, + "fallbacks": [], + } + await create(**request) + first: Final = logger._enqueue_deferred_logging + assert first is not None + logger._enqueue_deferred_logging = None + await asyncio.gather(*tuple(_PENDING_CACHE_WRITES)) + await create(**request) + cached: Final = await asyncio.wait_for(receipts.get(), 10) + assert cached["response_cost"] == 0 and requests.qsize() == 1 + assert await evaluation_spend_cache.async_get_cache("spend:user:creator") > 0 + await create(**{**request, "messages": [{"role": "user", "content": "two"}]}) + second: Final = logger._enqueue_deferred_logging + assert second is not None + first() + second() + received: Final = (await asyncio.wait_for(receipts.get(), 10), await asyncio.wait_for(receipts.get(), 10)) + await GLOBAL_LOGGING_WORKER.flush() + assert requests.qsize() == 2 and receipts.empty() + assert {receipt["response_cost"] for receipt in received} == {10 * 0.001 + 0.002, 10 * 0.001 + 2 * 0.002} + assert tuple(receipt["user"] for receipt in received) == ("creator", "creator") + assert await evaluation_spend_cache.async_get_cache("spend:user:creator") == pytest.approx(20 * 0.001 + 3 * 0.002) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("api", ("chat", "messages")) +async def test_cancelled_evaluation_releases_output_hold_and_keeps_input_cost( + evaluation_spend_cache: DualCache, evaluation_model: str, api: str +) -> None: + entered: Final = asyncio.Event() + owner: Final = EvaluationBillingOwner("creator", {evaluation_model: {"max_budget": 1, "budget_duration": "1d"}}, 1) + + async def upstream(request: httpx.Request) -> httpx.Response: + entered.set() + return await asyncio.Future[httpx.Response]() + + with respx.mock(assert_all_called=False) as transport, evaluation_billing_context(owner): + transport.post("https://evaluation.invalid/v1/chat/completions").mock(side_effect=upstream) + create: Final = litellm.acompletion if api == "chat" else litellm.anthropic.messages.acreate + pending: Final = asyncio.create_task( + create( + model=evaluation_model, + messages=[{"role": "user", "content": "hello"}], + max_tokens=10, + api_base="https://evaluation.invalid/v1", + api_key="test", + fallbacks=[], + num_retries=0, + ) + ) + await asyncio.wait_for(entered.wait(), 5) + held: Final = await evaluation_spend_cache.async_get_cache("spend:user:creator") + pending.cancel() + with pytest.raises(asyncio.CancelledError): + await pending + settled: Final = await evaluation_spend_cache.async_get_cache("spend:user:creator") + assert 0 < settled < held + assert await evaluation_spend_cache.async_get_cache(f"user_model_spend:creator:{evaluation_model}:1d") == settled + + +@pytest.mark.asyncio +@pytest.mark.parametrize("api", ("chat", "messages")) +@pytest.mark.parametrize("intercept_stream", (False, True)) +async def test_evaluation_streams_are_rejected_before_paid_transport( + evaluation_spend_cache: DualCache, + evaluation_model: str, + monkeypatch: pytest.MonkeyPatch, + api: str, + intercept_stream: bool, +) -> None: + from litellm.integrations.code_interpreter_interception.handler import CodeInterpreterInterceptionLogger + + monkeypatch.setattr(litellm, "callbacks", [CodeInterpreterInterceptionLogger()] if intercept_stream else []) + create: Final = litellm.acompletion if api == "chat" else litellm.anthropic.messages.acreate + with ( + respx.mock(assert_all_called=False) as transport, + evaluation_billing_context(EvaluationBillingOwner("creator", max_budget=1)), + ): + upstream: Final = transport.post("https://evaluation.invalid/v1/chat/completions").respond(500) + with pytest.raises(litellm.BadRequestError, match="Shadow evaluation requires a non-streaming response"): + await create( + model=evaluation_model, + messages=[{"role": "user", "content": "hello"}], + max_tokens=10, + api_base="https://evaluation.invalid/v1", + api_key="test", + fallbacks=[], + stream=True, + tools=[{"type": "code_interpreter"}], + ) + assert upstream.call_count == 0 + assert await evaluation_spend_cache.async_get_cache("spend:user:creator") is None + + +@pytest.mark.asyncio +async def test_internal_evaluation_streams_do_not_claim_budget( + evaluation_spend_cache: DualCache, evaluation_model: str +) -> None: + token: Final = is_internal_call.set(True) + try: + with evaluation_billing_context(EvaluationBillingOwner("creator", max_budget=0)): + response: Final = await litellm.acompletion( + model=evaluation_model, + messages=[{"role": "user", "content": "hello"}], + max_tokens=10, + stream=True, + mock_response="ok", + ) + assert isinstance(response, CustomStreamWrapper) + await response.aclose() + finally: + is_internal_call.reset(token) + assert await evaluation_spend_cache.async_get_cache("spend:user:creator") is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("budget", (0, 1)) +@pytest.mark.parametrize("api", ("chat", "messages")) +async def test_global_fallback_leaves_enforce_creator_budget_and_emit_one_paid_receipt( + evaluation_spend_cache: DualCache, evaluation_model: str, monkeypatch: pytest.MonkeyPatch, budget: int, api: str +) -> None: + fallback: Final = evaluation_model + "-fallback" + monkeypatch.setattr(litellm, "model_fallbacks", [fallback]) + receipts: Final[asyncio.Queue[Mapping[str, object]]] = asyncio.Queue() + owner: Final = EvaluationBillingOwner( + "creator", + {model: {"max_budget": 1, "budget_duration": "1d"} for model in (evaluation_model, fallback)}, + max_budget=budget, + ) + + async def capture(kwargs: Mapping[str, object], response: object, start: datetime, end: datetime) -> None: + receipts.put_nowait(kwargs) + + with ( + respx.mock(assert_all_called=False) as transport, + evaluation_billing_context(owner), + ): + upstream: Final = transport.post("https://evaluation.invalid/v1/chat/completions").mock( + side_effect=[ + httpx.Response(500, json={"error": {"message": "unavailable", "type": "server_error"}}), + httpx.Response( + 200, + json=ModelResponse( + model="evaluation-test-fallback", + choices=[{"message": {"role": "assistant", "content": "ok"}}], + usage=Usage(prompt_tokens=10, completion_tokens=2, total_tokens=12), + ).model_dump(), + ), + ], + ) + create: Final = litellm.acompletion if api == "chat" else litellm.anthropic.messages.acreate + request: Final = create( + model=evaluation_model, + messages=[{"role": "user", "content": "hello"}], + max_tokens=10, + api_base="https://evaluation.invalid/v1", + api_key="test", + num_retries=0, + max_retries=0, + success_callback=[capture], + ) + if budget == 0: + with pytest.raises(Exception, match="Max budget: 0"): + await request + else: + await request + receipt: Final = await asyncio.wait_for(receipts.get(), 10) + assert (receipt["user"], receipt["response_cost"]) == ("creator", 10 * 0.001 + 2 * 0.002) + await GLOBAL_LOGGING_WORKER.flush() + assert upstream.call_count == budget * 2 + assert receipts.empty() + assert (await evaluation_spend_cache.async_get_cache("spend:user:creator") or 0) == pytest.approx(budget * 0.014) + assert (await evaluation_spend_cache.async_get_cache(f"user_model_spend:creator:{evaluation_model}:1d") or 0) == 0 + assert ( + await evaluation_spend_cache.async_get_cache(f"user_model_spend:creator:{fallback}:1d") or 0 + ) == pytest.approx(budget * 0.014) diff --git a/ui/litellm-dashboard/src/app/(dashboard)/cost-optimization/_components/ShadowEvalStartForm.tsx b/ui/litellm-dashboard/src/app/(dashboard)/cost-optimization/_components/ShadowEvalStartForm.tsx index 06e332d2cfc..b332fa1ad91 100644 --- a/ui/litellm-dashboard/src/app/(dashboard)/cost-optimization/_components/ShadowEvalStartForm.tsx +++ b/ui/litellm-dashboard/src/app/(dashboard)/cost-optimization/_components/ShadowEvalStartForm.tsx @@ -38,9 +38,9 @@ const DIRECTION_OPTIONS: readonly { value: ShadowEvalDirection; label: string }[ const START_FORM_DESCRIPTION: Record = { forward: - "Duplicates a sampled slice of the selected targets' traffic (keys, teams, or users) through the auto-router and has an LLM judge compare both answers blind. Each target gets its own spend budget. The router's answers are never served to users; judge calls bill to the sampled traffic's own identity.", + "Duplicates a sampled slice of the selected targets' traffic (keys, teams, or users) through the auto-router and has an LLM judge compare both answers blind. Each target gets its own spend budget. The router's answers are never served to users; shadow and judge calls bill to the initiating admin.", reverse: - "Duplicates a sampled slice of the traffic the auto-router already serves against a fixed baseline model and has an LLM judge compare both answers blind. Each target gets its own spend budget. The baseline's answers are never served to users; judge calls bill to the sampled traffic's own identity.", + "Duplicates a sampled slice of the traffic the auto-router already serves against a fixed baseline model and has an LLM judge compare both answers blind. Each target gets its own spend budget. The baseline's answers are never served to users; shadow and judge calls bill to the initiating admin.", }; const DURATION_OPTIONS = [ diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index 3ed4103a227..27d061d78a3 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -1475,7 +1475,7 @@ export interface paths { * eval spend, the shadow and judge calls' own cost, reaches max_budget dollars, the * job's window ends, or the job is stopped, so one target running out of budget does * not end sampling for the others; sampling changes propagate to pods within about 10 - * seconds. Shadow and judge calls bill to the sampled request's own identity but are + * seconds. Shadow and judge calls bill to the admin who started the job and are * excluded from request counts and auto-router adoption metrics. */ post: operations["start_shadow_eval_auto_router_shadow_eval_start_post"];