Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion litellm/caching/caching_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -738,10 +738,13 @@ def _async_log_cache_hit_on_callbacks(
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(
Comment thread
tin-berri marked this conversation as resolved.
result=cached_result,
start_time=start_time,
end_time=end_time,
Expand Down
2 changes: 1 addition & 1 deletion litellm/caching/in_memory_cache.py
Original file line number Diff line number Diff line change
Expand Up @@ -168,7 +168,7 @@ def set_cache(self, key, value, **kwargs):
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))
Expand Down
12 changes: 11 additions & 1 deletion litellm/integrations/lago.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -61,6 +66,9 @@ def validate_environment(self):
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)
Expand Down Expand Up @@ -96,7 +104,9 @@ def _common_logic(self, kwargs: dict, response_obj) -> dict:
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
Expand Down
76 changes: 59 additions & 17 deletions litellm/integrations/shadow_eval_logger.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,11 @@
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,
Expand Down Expand Up @@ -76,6 +80,7 @@
_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], ...]:
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -1082,22 +1088,27 @@ async def _run_shadow_eval(
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,
Expand Down Expand Up @@ -1285,6 +1296,7 @@ async def _call_router_shadow(
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,
Expand Down Expand Up @@ -1348,6 +1360,7 @@ async def _call_judge(
judge_model,
judge_messages, # pyright: ignore[reportArgumentType] # plain SDK message dicts
team_id=_forwarded_team_id(parent_metadata),
stream=False,
Comment thread
cursor[bot] marked this conversation as resolved.
temperature=0,
max_tokens=JUDGE_MAX_OUTPUT_TOKENS,
response_format=PAIRWISE_JUDGE_RESPONSE_FORMAT,
Expand Down Expand Up @@ -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,
)
2 changes: 1 addition & 1 deletion litellm/litellm_core_utils/fallback_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)
Expand Down
110 changes: 105 additions & 5 deletions litellm/litellm_core_utils/internal_call_metadata.py
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -17,16 +15,118 @@

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.

Expand Down
Loading
Loading