add openrouter style cost details

This commit is contained in:
ali-corpo 2026-05-22 13:40:10 +00:00
parent d04373f4ce
commit 24026c1fd9
4 changed files with 302 additions and 0 deletions

View file

@ -320,6 +320,7 @@ banned_keywords_list: Optional[Union[str, List]] = None
llm_guard_mode: Literal["all", "key-specific", "request-specific"] = "all"
guardrail_name_config_map: Dict[str, GuardrailItem] = {}
include_cost_in_streaming_usage: bool = False
include_openrouter_style_cost_details: bool = False
reasoning_auto_summary: bool = False
### PROMPTS ####
from litellm.types.prompts.init_prompts import PromptSpec

View file

@ -981,6 +981,148 @@ class ProxyBaseLLMRequestProcessing:
model_id = model_info.get("id", "") or ""
return model_id
@staticmethod
def _resolve_is_byok_flag(
data: dict,
logging_obj: Optional[LiteLLMLoggingObj] = None,
) -> bool:
"""Resolve BYOK status from request metadata or logging context."""
litellm_metadata = data.get("litellm_metadata", {}) or {}
model_info = litellm_metadata.get("model_info", {}) or {}
if isinstance(model_info, dict) and isinstance(model_info.get("is_byok"), bool):
return model_info["is_byok"]
if logging_obj is not None and hasattr(logging_obj, "litellm_params"):
litellm_params = logging_obj.litellm_params or {}
direct_model_info = litellm_params.get("model_info", {}) or {}
if isinstance(direct_model_info, dict) and isinstance(
direct_model_info.get("is_byok"), bool
):
return direct_model_info["is_byok"]
nested_metadata = litellm_params.get("metadata", {}) or {}
nested_model_info = nested_metadata.get("model_info", {}) or {}
if isinstance(nested_model_info, dict) and isinstance(
nested_model_info.get("is_byok"), bool
):
return nested_model_info["is_byok"]
return False
@staticmethod
def _add_openrouter_style_usage_fields(
response: Any,
model_name: str,
is_byok: bool,
response_cost: Optional[Union[float, str]],
response_cost_details: Optional[dict],
) -> None:
"""Add OpenRouter-style usage metadata (`cost`, `is_byok`, `cost_details`)."""
if not getattr(litellm, "include_openrouter_style_cost_details", False):
return
if response is None:
return
usage_obj: Optional[Union[dict, Usage]] = None
if isinstance(response, dict):
usage_obj = response.get("usage")
else:
usage_obj = getattr(response, "usage", None)
if not isinstance(usage_obj, (dict, Usage)):
return
def _get_usage_value(key: str, default: Any = None) -> Any:
if isinstance(usage_obj, dict):
return usage_obj.get(key, default)
return getattr(usage_obj, key, default)
def _set_usage_value(key: str, value: Any) -> None:
if isinstance(usage_obj, dict):
usage_obj[key] = value
else:
setattr(usage_obj, key, value)
parsed_response_cost: Optional[float] = None
if response_cost not in (None, ""):
try:
parsed_response_cost = float(response_cost)
except (TypeError, ValueError):
parsed_response_cost = None
prompt_tokens = int(
_get_usage_value("prompt_tokens", _get_usage_value("input_tokens", 0)) or 0
)
completion_tokens = int(
_get_usage_value(
"completion_tokens", _get_usage_value("output_tokens", 0)
)
or 0
)
total_tokens = int(
_get_usage_value("total_tokens", prompt_tokens + completion_tokens)
or (prompt_tokens + completion_tokens)
)
prompt_cost: Optional[float] = None
completion_cost: Optional[float] = None
if model_name:
try:
prompt_cost, completion_cost = litellm.cost_per_token(
model=model_name,
usage=Usage(
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=total_tokens,
prompt_tokens_details=_get_usage_value(
"prompt_tokens_details", None
),
completion_tokens_details=_get_usage_value(
"completion_tokens_details", None
),
),
)
except Exception:
prompt_cost, completion_cost = None, None
calculated_total_cost: Optional[float] = None
if prompt_cost is not None and completion_cost is not None:
calculated_total_cost = float(prompt_cost + completion_cost)
final_cost = (
parsed_response_cost
if parsed_response_cost is not None
else calculated_total_cost
)
if final_cost is not None:
_set_usage_value("cost", final_cost)
_set_usage_value("is_byok", is_byok)
existing_cost_details = _get_usage_value("cost_details", None)
merged_cost_details: Dict[str, float] = {}
if isinstance(existing_cost_details, dict):
merged_cost_details.update(existing_cost_details)
if isinstance(response_cost_details, dict) and response_cost_details:
merged_cost_details.update(response_cost_details)
else:
if final_cost is not None:
merged_cost_details["upstream_inference_cost"] = float(final_cost)
if prompt_cost is not None:
merged_cost_details["upstream_inference_prompt_cost"] = float(
prompt_cost
)
if completion_cost is not None:
merged_cost_details["upstream_inference_completions_cost"] = float(
completion_cost
)
if merged_cost_details:
_set_usage_value("cost_details", merged_cost_details)
def _debug_log_request_payload(self) -> None:
"""Log request payload at DEBUG level, truncating if too large."""
if not verbose_proxy_logger.isEnabledFor(logging.DEBUG):
@ -1427,6 +1569,18 @@ class ProxyBaseLLMRequestProcessing:
# Always return the client-requested model name (not provider-prefixed internal identifiers)
# for OpenAI-compatible responses.
hidden_params_after_hooks = getattr(response, "_hidden_params", {}) or {}
ProxyBaseLLMRequestProcessing._add_openrouter_style_usage_fields(
response=response,
model_name=str(self.data.get("model") or getattr(response, "model", "") or ""),
is_byok=ProxyBaseLLMRequestProcessing._resolve_is_byok_flag(
data=self.data,
logging_obj=logging_obj,
),
response_cost=hidden_params_after_hooks.get("response_cost"),
response_cost_details=hidden_params_after_hooks.get("response_cost_details"),
)
if requested_model_from_client:
_override_openai_response_model(
response_obj=response,

View file

@ -1542,6 +1542,8 @@ class Usage(SafeAttributeModel, CompletionUsage):
server_tool_use: Optional[ServerToolUse] = None
cost: Optional[float] = None
is_byok: Optional[bool] = None
cost_details: Optional[Dict[str, float]] = None
completion_tokens_details: Optional[CompletionTokensDetailsWrapper] = None
"""Breakdown of tokens used in a completion."""
@ -1563,6 +1565,8 @@ class Usage(SafeAttributeModel, CompletionUsage):
] = None,
server_tool_use: Optional[ServerToolUse] = None,
cost: Optional[float] = None,
is_byok: Optional[bool] = None,
cost_details: Optional[Dict[str, float]] = None,
**params,
):
# handle reasoning_tokens
@ -1672,6 +1676,16 @@ class Usage(SafeAttributeModel, CompletionUsage):
else:
del self.cost
if is_byok is not None:
self.is_byok = is_byok
else:
del self.is_byok
if cost_details is not None:
self.cost_details = cost_details
else:
del self.cost_details
## ANTHROPIC MAPPING ##
if "cache_creation_input_tokens" in params and isinstance(
params["cache_creation_input_tokens"], int

View file

@ -942,6 +942,139 @@ class TestProxyBaseLLMRequestProcessing:
), f"queue_time_seconds should be at least 0.5, got {metadata['queue_time_seconds']}"
def test_add_openrouter_style_usage_fields_with_explicit_cost_details():
original_flag = getattr(litellm, "include_openrouter_style_cost_details", False)
litellm.include_openrouter_style_cost_details = True
response_obj = {
"usage": {
"prompt_tokens": 12000,
"completion_tokens": 452,
"total_tokens": 12452,
}
}
try:
ProxyBaseLLMRequestProcessing._add_openrouter_style_usage_fields(
response=response_obj,
model_name="openai/gpt-4o-mini",
is_byok=False,
response_cost=0.007356,
response_cost_details={
"upstream_inference_cost": 0.007356,
"upstream_inference_prompt_cost": 0.006,
"upstream_inference_completions_cost": 0.001356,
},
)
usage = response_obj["usage"]
assert usage["cost"] == pytest.approx(0.007356)
assert usage["is_byok"] is False
assert usage["cost_details"]["upstream_inference_cost"] == pytest.approx(
0.007356
)
assert usage["cost_details"][
"upstream_inference_prompt_cost"
] == pytest.approx(0.006)
assert usage["cost_details"][
"upstream_inference_completions_cost"
] == pytest.approx(0.001356)
finally:
litellm.include_openrouter_style_cost_details = original_flag
def test_add_openrouter_style_usage_fields_computes_prompt_and_completion_costs(
monkeypatch,
):
original_flag = getattr(litellm, "include_openrouter_style_cost_details", False)
litellm.include_openrouter_style_cost_details = True
response_obj = {
"usage": {
"prompt_tokens": 12000,
"completion_tokens": 452,
"total_tokens": 12452,
}
}
monkeypatch.setattr(
litellm,
"cost_per_token",
lambda model, usage: (0.006, 0.001356),
)
try:
ProxyBaseLLMRequestProcessing._add_openrouter_style_usage_fields(
response=response_obj,
model_name="openai/gpt-4o-mini",
is_byok=True,
response_cost=0.007356,
response_cost_details=None,
)
usage = response_obj["usage"]
assert usage["cost"] == pytest.approx(0.007356)
assert usage["is_byok"] is True
assert usage["cost_details"]["upstream_inference_cost"] == pytest.approx(
0.007356
)
assert usage["cost_details"][
"upstream_inference_prompt_cost"
] == pytest.approx(0.006)
assert usage["cost_details"][
"upstream_inference_completions_cost"
] == pytest.approx(0.001356)
finally:
litellm.include_openrouter_style_cost_details = original_flag
def test_add_openrouter_style_usage_fields_is_noop_when_disabled():
original_flag = getattr(litellm, "include_openrouter_style_cost_details", False)
litellm.include_openrouter_style_cost_details = False
response_obj = {
"usage": {
"prompt_tokens": 12000,
"completion_tokens": 452,
"total_tokens": 12452,
}
}
try:
ProxyBaseLLMRequestProcessing._add_openrouter_style_usage_fields(
response=response_obj,
model_name="openai/gpt-4o-mini",
is_byok=False,
response_cost=0.007356,
response_cost_details={
"upstream_inference_cost": 0.007356,
},
)
usage = response_obj["usage"]
assert "cost" not in usage
assert "is_byok" not in usage
assert "cost_details" not in usage
finally:
litellm.include_openrouter_style_cost_details = original_flag
def test_resolve_is_byok_flag_prefers_request_metadata():
data = {"litellm_metadata": {"model_info": {"is_byok": True}}}
assert ProxyBaseLLMRequestProcessing._resolve_is_byok_flag(data=data) is True
data_without_flag = {"litellm_metadata": {"model_info": {}}}
logging_obj = MagicMock()
logging_obj.litellm_params = {"model_info": {"is_byok": False}}
assert (
ProxyBaseLLMRequestProcessing._resolve_is_byok_flag(
data=data_without_flag,
logging_obj=logging_obj,
)
is False
)
@pytest.mark.asyncio
class TestCommonRequestProcessingHelpers:
async def consume_stream(self, streaming_response: StreamingResponse) -> list: