scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-04 02:31:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 122

feat(proxy): add TypeSafe Jev passthrough spend tracking

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
mateo 2026-09-17 15:53:19 +00:00
parent 4b368bf066
commit 2dc9697381
12 changed files with 523 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
litellm/model_prices_and_context_window_backup.json
View file

@ -69158,5 +69158,26 @@
"supports_prompt_caching": true,
"supports_reasoning": true,
"supports_vision": true
},
"typesafe/jev-1.13.0": {
"input_cost_per_token": 4.2e-08,
"litellm_provider": "typesafe",
"mode": "evaluation",
"output_cost_per_token": 0.0,
"source": "https://docs.typesafe.ai/models"
},
"typesafe/jev-latest": {
"input_cost_per_token": 4.2e-08,
"litellm_provider": "typesafe",
"mode": "evaluation",
"output_cost_per_token": 0.0,
"source": "https://docs.typesafe.ai/models"
},
"typesafe/jev-preview": {
"input_cost_per_token": 4.2e-08,
"litellm_provider": "typesafe",
"mode": "evaluation",
"output_cost_per_token": 0.0,
"source": "https://docs.typesafe.ai/models"
}
}

1
litellm/proxy/_lazy_features.py
View file

@ -208,6 +208,7 @@ LAZY_FEATURES: Final[tuple[LazyFeature, ...]] = (
"/nvidia_nim/",
"/openai/",
"/openai_passthrough/",
"/typesafe/",
"/vertex-ai/",
"/vertex_ai/",
"/vllm/",

90
litellm/proxy/_lazy_openapi_snapshot.json
View file

@ -20373,6 +20373,96 @@
]
}
},
"/typesafe/{endpoint}": {
"get": {
"description": "[Docs](https://docs.litellm.ai/docs/pass_through/typesafe)",
"operationId": "typesafe_proxy_route_typesafe__endpoint__get",
"parameters": [
{
"in": "path",
"name": "endpoint",
"required": true,
"schema": {
"title": "Endpoint",
"type": "string"
}
}
],
"responses": {
"200": {
"content": {
"application/json": {
"schema": {}
}
},
"description": "Successful Response"
},
"422": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
"description": "Validation Error"
}
},
"security": [
{
"APIKeyHeader": []
}
],
"summary": "Typesafe Proxy Route",
"tags": [
"llm_passthrough"
]
},
"post": {
"description": "[Docs](https://docs.litellm.ai/docs/pass_through/typesafe)",
"operationId": "typesafe_proxy_route_typesafe__endpoint__post",
"parameters": [
{
"in": "path",
"name": "endpoint",
"required": true,
"schema": {
"title": "Endpoint",
"type": "string"
}
}
],
"responses": {
"200": {
"content": {
"application/json": {
"schema": {}
}
},
"description": "Successful Response"
},
"422": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
"description": "Validation Error"
}
},
"security": [
{
"APIKeyHeader": []
}
],
"summary": "Typesafe Proxy Route",
"tags": [
"llm_passthrough"
]
}
},
"/vertex_ai/discovery/{endpoint}": {
"delete": {
"description": "Call any vertex discovery endpoint using the proxy.\n\nJust use `{PROXY_BASE_URL}/vertex_ai/discovery/{endpoint:path}`\n\nTarget url: `https://discoveryengine.googleapis.com`",

1
litellm/proxy/_types.py
View file

@ -483,6 +483,7 @@ class LiteLLMRoutes(enum.Enum):
"/eu.assemblyai",
"/vllm",
"/mistral",
"/typesafe",
"/milvus",
"/gigachat",
"/watsonx",

47
litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py
View file

@ -525,6 +525,53 @@ async def mistral_proxy_route(
return received_value
@router.api_route(
"/typesafe/{endpoint:path}",
methods=["GET", "POST"],
tags=["TypeSafe AI Pass-through", "pass-through"],
)
async def typesafe_proxy_route(
endpoint: str,
request: Request,
fastapi_response: Response,
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
"""[Docs](https://docs.litellm.ai/docs/pass_through/typesafe)"""
if request.method == "POST":
try:
request_body: Final = await _json_request_body(request)
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
if not isinstance(request_body, dict):
raise HTTPException(status_code=400, detail="Request body must be a JSON object")
if "stream" in request_body:
raise HTTPException(status_code=400, detail="'stream' is not a TypeSafe request member")
base_target_url: Final = get_secret_str("TYPESAFE_API_BASE") or "https://api.typesafe.ai"
encoded_endpoint: Final = httpx.URL(endpoint).path
normalized_endpoint: Final = encoded_endpoint if encoded_endpoint.startswith("/") else f"/{encoded_endpoint}"
base_url: Final = httpx.URL(base_target_url)
updated_url: Final = base_url.copy_with(
path=HttpPassThroughEndpointHelpers.join_base_and_endpoint_path(base_url, normalized_endpoint),
params=request.query_params,
)
typesafe_api_key: Final = passthrough_endpoint_router.get_credentials(
custom_llm_provider="typesafe",
region_name=None,
)
endpoint_func: Final = create_pass_through_route(
endpoint=endpoint,
target=str(updated_url),
custom_headers={
"Authorization": f"Bearer {typesafe_api_key}",
"Content-Type": "application/json",
},
custom_llm_provider="typesafe",
is_streaming_request=False,
)
return await endpoint_func(request, fastapi_response, user_api_key_dict)
@router.api_route(
"/milvus/{endpoint:path}",
methods=["GET", "POST", "PUT", "DELETE", "PATCH"],

116
litellm/proxy/pass_through_endpoints/llm_provider_handlers/typesafe_passthrough_logging_handler.py Normal file
View file

@ -0,0 +1,116 @@
from collections.abc import Mapping
from datetime import datetime
from typing import Final, cast
import httpx
from pydantic import BaseModel, TypeAdapter, ValidationError
import litellm
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.litellm_logging import (
get_standard_logging_object_payload, # pyright: ignore[reportUnknownVariableType] # legacy helper has an untyped signature
)
from litellm.proxy._types import PassThroughEndpointLoggingTypedDict
from litellm.types.utils import ModelResponse, StandardPassThroughResponseObject, Usage
class _TypeSafeUsage(BaseModel):
input_tokens: int = 0
output_tokens: int = 0
class _TypeSafeResponse(BaseModel):
model: str | None = None
usage: _TypeSafeUsage | None = None
_TYPESAFE_RESPONSE_ADAPTER: Final = TypeAdapter(_TypeSafeResponse)
_MODEL_COST_ENTRY_ADAPTER: Final = TypeAdapter(dict[str, object])
def _parse_typesafe_response(response_body: Mapping[str, object]) -> _TypeSafeResponse:
try:
return _TYPESAFE_RESPONSE_ADAPTER.validate_python(response_body)
except ValidationError:
return _TypeSafeResponse()
def _get_model_cost_entry(model_key: str) -> Mapping[str, object] | None:
model_cost: Final[Mapping[str, object]] = cast(Mapping[str, object], litellm.model_cost)
entry: Final[object] = model_cost.get(model_key)
try:
return _MODEL_COST_ENTRY_ADAPTER.validate_python(entry)
except ValidationError:
return None
class TypeSafePassthroughLoggingHandler:
@staticmethod
def typesafe_passthrough_handler(
httpx_response: httpx.Response,
response_body: Mapping[str, object],
logging_obj: LiteLLMLoggingObj,
url_route: str,
result: str,
start_time: datetime,
end_time: datetime,
cache_hit: bool,
request_body: Mapping[str, object],
**kwargs: object,
) -> PassThroughEndpointLoggingTypedDict:
response: Final = _parse_typesafe_response(response_body)
response_model: Final = response.model
request_model_value: Final = request_body.get("model")
request_model: Final = request_model_value if isinstance(request_model_value, str) else None
logged_model: Final = response_model or request_model or "jev-latest"
model_name: Final = f"typesafe/{logged_model}"
usage: Final = response.usage or _TypeSafeUsage()
input_tokens: Final = usage.input_tokens
output_tokens: Final = usage.output_tokens
candidate_model_keys: Final = tuple(
f"typesafe/{model}" for model in (response_model, request_model) if model is not None
)
cost_entry: Final = next(
(entry for model_key in candidate_model_keys if (entry := _get_model_cost_entry(model_key)) is not None),
None,
)
input_cost_per_token: Final = (
cost_entry.get("input_cost_per_token", 0.0) if isinstance(cost_entry, Mapping) else 0.0
)
output_cost_per_token: Final = (
cost_entry.get("output_cost_per_token", 0.0) if isinstance(cost_entry, Mapping) else 0.0
)
response_cost: Final = (
input_tokens * float(input_cost_per_token) + output_tokens * float(output_cost_per_token)
if isinstance(input_cost_per_token, (int, float)) and isinstance(output_cost_per_token, (int, float))
else 0.0
)
usage_object: Final = Usage(
prompt_tokens=input_tokens,
completion_tokens=output_tokens,
total_tokens=input_tokens + output_tokens,
)
updated_kwargs: Final = {
**kwargs,
"model": model_name,
"custom_llm_provider": "typesafe",
"response_cost": response_cost,
"combined_usage_object": usage_object,
}
logging_obj.model_call_details.update(
model=model_name,
custom_llm_provider="typesafe",
response_cost=response_cost,
)
standard_logging_object: Final = get_standard_logging_object_payload(
kwargs=updated_kwargs,
init_response_obj=ModelResponse(model=model_name, usage=usage_object),
start_time=start_time,
end_time=end_time,
logging_obj=logging_obj,
status="success",
)
return {
"result": StandardPassThroughResponseObject(response=result),
"kwargs": {**updated_kwargs, "standard_logging_object": standard_logging_object},
}

22
litellm/proxy/pass_through_endpoints/success_handler.py
View file

@ -256,6 +256,25 @@ class PassThroughEndpointLogging:
)
standard_logging_response_object = comprehend_medical_handler_result["result"] # rebind-ok: elif-chain
kwargs = comprehend_medical_handler_result["kwargs"] # rebind-ok: elif-chain contract
elif self.is_typesafe_route(custom_llm_provider):
from .llm_provider_handlers.typesafe_passthrough_logging_handler import (
TypeSafePassthroughLoggingHandler,
)
typesafe_handler_result: Final = TypeSafePassthroughLoggingHandler.typesafe_passthrough_handler(
httpx_response=httpx_response,
response_body=response_body if isinstance(response_body, dict) else {},
logging_obj=logging_obj,
url_route=url_route,
result=result,
start_time=start_time,
end_time=end_time,
cache_hit=cache_hit,
request_body=request_body,
**kwargs,
)
standard_logging_response_object = typesafe_handler_result["result"]
kwargs = typesafe_handler_result["kwargs"]
elif self.is_vertex_ai_live_route(url_route):
from .llm_provider_handlers.vertex_ai_live_passthrough_logging_handler import (
VertexAILivePassthroughLoggingHandler,
@ -389,6 +408,9 @@ class PassThroughEndpointLogging:
def is_comprehend_medical_route(self, custom_llm_provider: str | None) -> bool:
return custom_llm_provider == "comprehendmedical"
def is_typesafe_route(self, custom_llm_provider: str | None) -> bool:
return custom_llm_provider == "typesafe"
def is_langfuse_route(self, url_route: str):
parsed_url: Final = urlparse(url_route)
for route in self.TRACKED_LANGFUSE_ROUTES:

1
litellm/types/utils.py
View file

@ -348,6 +348,7 @@ class ModelInfoBase(ProviderSpecificModelInfo, total=False):
"audio_transcription",
"audio_speech",
"responses",
"evaluation",
"ocr",
"realtime",
]

21
model_prices_and_context_window.json
View file

@ -69158,5 +69158,26 @@
"supports_prompt_caching": true,
"supports_reasoning": true,
"supports_vision": true
},
"typesafe/jev-1.13.0": {
"input_cost_per_token": 4.2e-08,
"litellm_provider": "typesafe",
"mode": "evaluation",
"output_cost_per_token": 0.0,
"source": "https://docs.typesafe.ai/models"
},
"typesafe/jev-latest": {
"input_cost_per_token": 4.2e-08,
"litellm_provider": "typesafe",
"mode": "evaluation",
"output_cost_per_token": 0.0,
"source": "https://docs.typesafe.ai/models"
},
"typesafe/jev-preview": {
"input_cost_per_token": 4.2e-08,
"litellm_provider": "typesafe",
"mode": "evaluation",
"output_cost_per_token": 0.0,
"source": "https://docs.typesafe.ai/models"
}
}

127
tests/test_litellm/proxy/pass_through_endpoints/llm_provider_handlers/test_typesafe_passthrough_logging_handler.py Normal file
View file

@ -0,0 +1,127 @@
from datetime import datetime
from unittest.mock import MagicMock
import httpx
import pytest
import litellm
from litellm.proxy.pass_through_endpoints.llm_provider_handlers.typesafe_passthrough_logging_handler import (
TypeSafePassthroughLoggingHandler,
)
from litellm.proxy.pass_through_endpoints.success_handler import PassThroughEndpointLogging
@pytest.fixture(autouse=True)
def local_model_cost_map(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
def _response() -> httpx.Response:
return httpx.Response(
200,
request=httpx.Request("POST", "https://api.typesafe.ai/v1/systemone"),
json={"model": "jev-1.13.0"},
)
def _logging_obj() -> MagicMock:
logging_obj = MagicMock()
logging_obj.model_call_details = {}
return logging_obj
def _handler_result(response_body: dict, request_body: dict) -> dict:
return TypeSafePassthroughLoggingHandler.typesafe_passthrough_handler(
httpx_response=_response(),
response_body=response_body,
logging_obj=_logging_obj(),
url_route="https://api.typesafe.ai/v1/systemone",
result='{"answers": {}}',
start_time=datetime.now(),
end_time=datetime.now(),
cache_hit=False,
request_body=request_body,
)
def test_uses_registry_pricing_and_standard_usage():
logging_obj = _logging_obj()
model_key = "typesafe/jev-1.13.0"
model_cost = litellm.model_cost[model_key]
response = TypeSafePassthroughLoggingHandler.typesafe_passthrough_handler(
httpx_response=_response(),
response_body={"model": "jev-1.13.0", "usage": {"input_tokens": 312, "output_tokens": 48}},
logging_obj=logging_obj,
url_route="https://api.typesafe.ai/v1/systemone",
result='{"answers": {}}',
start_time=datetime.now(),
end_time=datetime.now(),
cache_hit=False,
request_body={"model": "jev-latest"},
)
expected_cost = 312 * model_cost["input_cost_per_token"] + 48 * model_cost["output_cost_per_token"]
assert response["kwargs"]["response_cost"] == pytest.approx(expected_cost)
assert response["kwargs"]["combined_usage_object"].prompt_tokens == 312
assert response["kwargs"]["combined_usage_object"].completion_tokens == 48
assert response["kwargs"]["combined_usage_object"].total_tokens == 360
def test_falls_back_to_request_model_when_response_model_is_missing():
result = _handler_result(
{"usage": {"input_tokens": 10, "output_tokens": 2}},
{"model": "jev-latest"},
)
model_cost = litellm.model_cost["typesafe/jev-latest"]
expected_cost = 10 * model_cost["input_cost_per_token"] + 2 * model_cost["output_cost_per_token"]
assert result["kwargs"]["model"] == "typesafe/jev-latest"
assert result["kwargs"]["response_cost"] == pytest.approx(expected_cost)
def test_missing_usage_is_zero_cost():
result = _handler_result({"model": "jev-1.13.0"}, {"model": "jev-latest"})
assert result["kwargs"]["response_cost"] == 0.0
def test_records_model_provider_and_cost_on_logging_details():
logging_obj = _logging_obj()
result = TypeSafePassthroughLoggingHandler.typesafe_passthrough_handler(
httpx_response=_response(),
response_body={"model": "jev-1.13.0", "usage": {"input_tokens": 1, "output_tokens": 0}},
logging_obj=logging_obj,
url_route="https://api.typesafe.ai/v1/systemone",
result="{}",
start_time=datetime.now(),
end_time=datetime.now(),
cache_hit=False,
request_body={"model": "jev-latest"},
)
assert result["kwargs"]["model"] == "typesafe/jev-1.13.0"
assert result["kwargs"]["custom_llm_provider"] == "typesafe"
assert result["kwargs"]["response_cost"] > 0
assert logging_obj.model_call_details["model"] == "typesafe/jev-1.13.0"
assert logging_obj.model_call_details["custom_llm_provider"] == "typesafe"
assert logging_obj.model_call_details["response_cost"] == result["kwargs"]["response_cost"]
def test_success_handler_dispatches_to_typesafe_handler():
logging_obj = _logging_obj()
normalized = PassThroughEndpointLogging().normalize_llm_passthrough_logging_payload(
httpx_response=_response(),
response_body={"model": "jev-1.13.0", "usage": {"input_tokens": 1, "output_tokens": 0}},
request_body={"model": "jev-latest"},
logging_obj=logging_obj,
url_route="https://api.typesafe.ai/v1/systemone",
result="{}",
start_time=datetime.now(),
end_time=datetime.now(),
cache_hit=False,
custom_llm_provider="typesafe",
)
assert normalized["kwargs"]["custom_llm_provider"] == "typesafe"
assert normalized["kwargs"]["model"] == "typesafe/jev-1.13.0"

59
tests/test_litellm/proxy/pass_through_endpoints/test_llm_pass_through_endpoints.py
View file

@ -43,6 +43,7 @@ from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import (
mistral_proxy_route,
relay_nvidia_nim_request,
openai_proxy_route,
typesafe_proxy_route,
vertex_discovery_proxy_route,
vertex_proxy_route,
vllm_proxy_route,
@ -6136,3 +6137,61 @@ class TestAzureRelayDeploymentSegment:
)
assert [call["model"] for call in captured] == ["gpt", "gpt"]
class TestTypeSafePassthroughRoute:
@staticmethod
def _request(body: object, query_params: Mapping[str, str] | None = None) -> MagicMock:
request = MagicMock(spec=Request)
request.method = "POST"
request.query_params = query_params or {}
request.json = AsyncMock(return_value=body)
return request
@pytest.mark.asyncio
async def test_forwards_target_auth_headers_provider_and_query(self, monkeypatch):
monkeypatch.setenv("TYPESAFE_API_KEY", "typesafe-test-key")
monkeypatch.setenv("TYPESAFE_API_BASE", "https://typesafe.example/base")
endpoint_func = AsyncMock(return_value={"ok": True})
create_route = Mock(return_value=endpoint_func)
monkeypatch.setattr(
"litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.create_pass_through_route",
create_route,
)
request = self._request({"state": "x"}, {"trace": "yes"})
result = await typesafe_proxy_route(
endpoint="v1/systemone",
request=request,
fastapi_response=MagicMock(spec=Response),
user_api_key_dict=UserAPIKeyAuth(api_key="virtual-key"),
)
assert result == {"ok": True}
endpoint_func.assert_awaited_once()
create_route.assert_called_once_with(
endpoint="v1/systemone",
target="https://typesafe.example/base/v1/systemone?trace=yes",
custom_headers={
"Authorization": "Bearer typesafe-test-key",
"Content-Type": "application/json",
},
custom_llm_provider="typesafe",
is_streaming_request=False,
)
assert request.json.await_count == 1
@pytest.mark.asyncio
async def test_rejects_stream_body(self, monkeypatch):
monkeypatch.setenv("TYPESAFE_API_KEY", "typesafe-test-key")
request = self._request({"stream": True})
with pytest.raises(HTTPException) as exc_info:
await typesafe_proxy_route(
endpoint="v1/systemone",
request=request,
fastapi_response=MagicMock(spec=Response),
user_api_key_dict=UserAPIKeyAuth(api_key="virtual-key"),
)
assert exc_info.value.status_code == 400

17
tests/test_litellm/test_typesafe_model_metadata.py Normal file
View file

@ -0,0 +1,17 @@
import pytest
import litellm
@pytest.fixture(autouse=True)
def local_model_cost_map(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
def test_typesafe_models_share_pricing_and_provider_metadata():
entries = [litellm.model_cost[f"typesafe/{model}"] for model in ("jev-1.13.0", "jev-latest", "jev-preview")]
assert {entry["input_cost_per_token"] for entry in entries} == {entries[0]["input_cost_per_token"]}
assert {entry["output_cost_per_token"] for entry in entries} == {entries[0]["output_cost_per_token"]}
assert {entry["litellm_provider"] for entry in entries} == {"typesafe"}