mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-05 02:41:56 +00:00
523 lines
22 KiB
Python
523 lines
22 KiB
Python
import json
|
|
from datetime import datetime
|
|
from types import MappingProxyType
|
|
from typing import Any, Final
|
|
from urllib.parse import urlparse
|
|
|
|
import httpx
|
|
|
|
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
|
|
from litellm.proxy._types import PassThroughEndpointLoggingResultValues
|
|
from litellm.types.passthrough_endpoints.pass_through_endpoints import (
|
|
PassthroughStandardLoggingPayload,
|
|
)
|
|
from litellm.types.utils import StandardPassThroughResponseObject
|
|
|
|
from .llm_provider_handlers.anthropic_passthrough_logging_handler import (
|
|
AnthropicPassthroughLoggingHandler,
|
|
)
|
|
from .llm_provider_handlers.assembly_passthrough_logging_handler import (
|
|
AssemblyAIPassthroughLoggingHandler,
|
|
)
|
|
from .llm_provider_handlers.cohere_passthrough_logging_handler import (
|
|
CoherePassthroughLoggingHandler,
|
|
)
|
|
from .llm_provider_handlers.cursor_passthrough_logging_handler import (
|
|
CursorPassthroughLoggingHandler,
|
|
)
|
|
from .llm_provider_handlers.gemini_passthrough_logging_handler import (
|
|
GeminiPassthroughLoggingHandler,
|
|
)
|
|
from .llm_provider_handlers.vertex_passthrough_logging_handler import (
|
|
VertexPassthroughLoggingHandler,
|
|
)
|
|
from .upstream_usage_headers import has_upstream_reported_usage
|
|
|
|
cohere_passthrough_logging_handler: Final = CoherePassthroughLoggingHandler()
|
|
|
|
|
|
def _safe_response_text(httpx_response: httpx.Response) -> str:
|
|
"""
|
|
Streamed passthrough responses are relayed to the client without being read
|
|
into memory, so accessing .text on them raises ResponseNotRead. Their body is
|
|
intentionally uninspected; log an empty string instead of failing the row.
|
|
"""
|
|
try:
|
|
return httpx_response.text
|
|
except httpx.ResponseNotRead:
|
|
return ""
|
|
|
|
|
|
class PassThroughEndpointLogging:
|
|
def __init__(self):
|
|
self.TRACKED_VERTEX_METHOD_ROUTES = (
|
|
"generateContent",
|
|
"streamGenerateContent",
|
|
"predict",
|
|
"rawPredict",
|
|
"streamRawPredict",
|
|
"search",
|
|
"predictLongRunning",
|
|
"embedContent",
|
|
"batchEmbedContents",
|
|
)
|
|
self.TRACKED_VERTEX_RESOURCE_ROUTES = ("batchPredictionJobs",)
|
|
|
|
# Anthropic
|
|
self.TRACKED_ANTHROPIC_ROUTES = ["/messages", "/v1/messages/batches"]
|
|
|
|
# Cohere
|
|
self.TRACKED_COHERE_ROUTES = ["/v2/chat", "/v1/embed"]
|
|
self.assemblyai_passthrough_logging_handler = AssemblyAIPassthroughLoggingHandler()
|
|
|
|
# Langfuse
|
|
self.TRACKED_LANGFUSE_ROUTES = ["/langfuse/"]
|
|
|
|
# Gemini
|
|
self.TRACKED_GEMINI_ROUTES = [
|
|
"generateContent",
|
|
"streamGenerateContent",
|
|
"predictLongRunning",
|
|
]
|
|
|
|
# Cursor Cloud Agents
|
|
self.TRACKED_CURSOR_ROUTES = [
|
|
"/v0/agents",
|
|
"/v0/me",
|
|
"/v0/models",
|
|
"/v0/repositories",
|
|
]
|
|
|
|
# Vertex AI Live API WebSocket
|
|
self.TRACKED_VERTEX_AI_LIVE_ROUTES = ["/vertex_ai/live"]
|
|
|
|
async def _handle_logging(
|
|
self,
|
|
logging_obj: LiteLLMLoggingObj,
|
|
standard_logging_response_object: StandardPassThroughResponseObject
|
|
| PassThroughEndpointLoggingResultValues
|
|
| dict,
|
|
result: str,
|
|
start_time: datetime,
|
|
end_time: datetime,
|
|
cache_hit: bool,
|
|
**kwargs,
|
|
):
|
|
"""Log pass-through success via the shared async dispatch path."""
|
|
# Always reached from pass_through_async_success_handler, which runs in
|
|
# an async context. call_type is "pass_through_endpoint" here, so the
|
|
# passthrough guard in dispatch_success_handlers already forces the
|
|
# async handler to run; pass prefer_async_handlers explicitly to match
|
|
# the streaming sibling (_route_streaming_logging_to_handler) and keep
|
|
# async-only loggers (e.g. the proxy spend logger) firing regardless of
|
|
# how the call-type classification evolves.
|
|
await logging_obj.dispatch_success_handlers(
|
|
result=(json.dumps(result) if isinstance(result, dict) else standard_logging_response_object),
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
cache_hit=False,
|
|
prefer_async_handlers=True,
|
|
**kwargs,
|
|
)
|
|
|
|
def normalize_llm_passthrough_logging_payload(
|
|
self,
|
|
httpx_response: httpx.Response,
|
|
response_body: dict | list[dict[str, object]] | None,
|
|
request_body: dict,
|
|
logging_obj: LiteLLMLoggingObj,
|
|
url_route: str,
|
|
result: str,
|
|
start_time: datetime,
|
|
end_time: datetime,
|
|
cache_hit: bool,
|
|
custom_llm_provider: str | None = None,
|
|
**kwargs,
|
|
):
|
|
return_dict: Final = {
|
|
"standard_logging_response_object": None,
|
|
"kwargs": kwargs,
|
|
}
|
|
standard_logging_response_object: Any | None = None
|
|
|
|
if self.is_gemini_route(url_route, custom_llm_provider):
|
|
gemini_passthrough_logging_handler_result = GeminiPassthroughLoggingHandler.gemini_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 = gemini_passthrough_logging_handler_result["result"]
|
|
kwargs = gemini_passthrough_logging_handler_result["kwargs"]
|
|
elif self.is_vertex_route(url_route):
|
|
vertex_passthrough_logging_handler_result = VertexPassthroughLoggingHandler.vertex_passthrough_handler(
|
|
httpx_response=httpx_response,
|
|
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 = vertex_passthrough_logging_handler_result["result"]
|
|
kwargs = vertex_passthrough_logging_handler_result["kwargs"]
|
|
elif self.is_anthropic_route(url_route):
|
|
anthropic_passthrough_logging_handler_result: Final = (
|
|
AnthropicPassthroughLoggingHandler.anthropic_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 = anthropic_passthrough_logging_handler_result["result"]
|
|
kwargs = anthropic_passthrough_logging_handler_result["kwargs"]
|
|
elif self.is_cohere_route(url_route):
|
|
cohere_passthrough_logging_handler_result = cohere_passthrough_logging_handler.cohere_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 = cohere_passthrough_logging_handler_result["result"]
|
|
kwargs = cohere_passthrough_logging_handler_result["kwargs"]
|
|
elif self.is_openai_route(url_route) and self._is_supported_openai_endpoint(url_route):
|
|
from .llm_provider_handlers.openai_passthrough_logging_handler import (
|
|
OpenAIPassthroughLoggingHandler,
|
|
)
|
|
|
|
openai_passthrough_logging_handler_result = OpenAIPassthroughLoggingHandler.openai_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 = openai_passthrough_logging_handler_result["result"]
|
|
kwargs = openai_passthrough_logging_handler_result["kwargs"]
|
|
|
|
elif self.is_cursor_route(url_route, custom_llm_provider):
|
|
cursor_passthrough_logging_handler_result = CursorPassthroughLoggingHandler.cursor_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 = cursor_passthrough_logging_handler_result["result"]
|
|
kwargs = cursor_passthrough_logging_handler_result["kwargs"]
|
|
elif self.is_comprehend_medical_route(custom_llm_provider):
|
|
from .llm_provider_handlers.comprehend_medical_passthrough_logging_handler import (
|
|
ComprehendMedicalPassthroughLoggingHandler,
|
|
)
|
|
|
|
comprehend_medical_handler_result: Final = (
|
|
ComprehendMedicalPassthroughLoggingHandler.comprehend_medical_passthrough_handler(
|
|
httpx_response=httpx_response,
|
|
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 = 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 MappingProxyType({}),
|
|
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,
|
|
)
|
|
|
|
vertex_ai_live_handler: Final = VertexAILivePassthroughLoggingHandler()
|
|
|
|
# For WebSocket responses, response_body should be a list of messages
|
|
websocket_messages: Final[list[dict[str, Any]]] = response_body if isinstance(response_body, list) else []
|
|
|
|
vertex_ai_live_handler_result: Final = vertex_ai_live_handler.vertex_ai_live_passthrough_handler(
|
|
websocket_messages=websocket_messages,
|
|
logging_obj=logging_obj,
|
|
url_route=url_route,
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
request_body=request_body,
|
|
**kwargs,
|
|
)
|
|
|
|
standard_logging_response_object = vertex_ai_live_handler_result["result"]
|
|
kwargs = vertex_ai_live_handler_result["kwargs"]
|
|
return_dict["standard_logging_response_object"] = standard_logging_response_object
|
|
|
|
return_dict["kwargs"] = kwargs
|
|
return return_dict
|
|
|
|
async def pass_through_async_success_handler(
|
|
self,
|
|
httpx_response: httpx.Response,
|
|
response_body: dict | list[dict[str, object]] | None,
|
|
logging_obj: LiteLLMLoggingObj,
|
|
url_route: str,
|
|
result: str,
|
|
start_time: datetime,
|
|
end_time: datetime,
|
|
cache_hit: bool,
|
|
request_body: dict,
|
|
passthrough_logging_payload: PassthroughStandardLoggingPayload,
|
|
custom_llm_provider: str | None = None,
|
|
**kwargs,
|
|
):
|
|
standard_logging_response_object: PassThroughEndpointLoggingResultValues | None = None
|
|
logging_obj.model_call_details["passthrough_logging_payload"] = passthrough_logging_payload
|
|
if self.is_assemblyai_route(url_route):
|
|
if AssemblyAIPassthroughLoggingHandler._should_log_request(httpx_response.request.method) is not True:
|
|
return
|
|
self.assemblyai_passthrough_logging_handler.assemblyai_passthrough_logging_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,
|
|
**kwargs,
|
|
)
|
|
return
|
|
elif self.is_langfuse_route(url_route):
|
|
# Don't log langfuse pass-through requests
|
|
return
|
|
else:
|
|
normalized_llm_passthrough_logging_payload: Final = self.normalize_llm_passthrough_logging_payload(
|
|
httpx_response=httpx_response,
|
|
response_body=response_body,
|
|
request_body=request_body,
|
|
logging_obj=logging_obj,
|
|
url_route=url_route,
|
|
result=result,
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
cache_hit=cache_hit,
|
|
custom_llm_provider=custom_llm_provider,
|
|
**kwargs,
|
|
)
|
|
standard_logging_response_object = normalized_llm_passthrough_logging_payload[
|
|
"standard_logging_response_object"
|
|
]
|
|
kwargs = normalized_llm_passthrough_logging_payload["kwargs"]
|
|
if standard_logging_response_object is None:
|
|
standard_logging_response_object = StandardPassThroughResponseObject(
|
|
response=_safe_response_text(httpx_response)
|
|
)
|
|
|
|
kwargs = self._set_cost_per_request(
|
|
logging_obj=logging_obj,
|
|
passthrough_logging_payload=passthrough_logging_payload,
|
|
kwargs=kwargs,
|
|
)
|
|
|
|
await self._handle_logging(
|
|
logging_obj=logging_obj,
|
|
standard_logging_response_object=standard_logging_response_object,
|
|
result=result,
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
cache_hit=cache_hit,
|
|
standard_pass_through_logging_payload=passthrough_logging_payload,
|
|
**kwargs,
|
|
)
|
|
|
|
def is_vertex_route(self, url_route: str) -> bool:
|
|
if any(f":{method}" in url_route for method in self.TRACKED_VERTEX_METHOD_ROUTES):
|
|
return True
|
|
if any(resource in url_route for resource in self.TRACKED_VERTEX_RESOURCE_ROUTES):
|
|
return True
|
|
return VertexPassthroughLoggingHandler.is_vertex_interactions_route(url_route)
|
|
|
|
def is_anthropic_route(self, url_route: str):
|
|
for route in self.TRACKED_ANTHROPIC_ROUTES:
|
|
if route in url_route:
|
|
return True
|
|
return False
|
|
|
|
def is_cohere_route(self, url_route: str) -> bool:
|
|
for route in self.TRACKED_COHERE_ROUTES:
|
|
if route not in url_route:
|
|
continue
|
|
if route == "/v1/embed" and "/v1/embeddings" in url_route:
|
|
continue
|
|
return True
|
|
return False
|
|
|
|
def is_assemblyai_route(self, url_route: str):
|
|
parsed_url: Final = urlparse(url_route)
|
|
if parsed_url.hostname == "api.assemblyai.com" or "/transcript" in parsed_url.path:
|
|
return True
|
|
return False
|
|
|
|
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:
|
|
if route in parsed_url.path:
|
|
return True
|
|
return False
|
|
|
|
def is_vertex_ai_live_route(self, url_route: str):
|
|
"""Check if the URL route is a Vertex AI Live API WebSocket route."""
|
|
if not url_route:
|
|
return False
|
|
for route in self.TRACKED_VERTEX_AI_LIVE_ROUTES:
|
|
if route in url_route:
|
|
return True
|
|
return False
|
|
|
|
def is_cursor_route(self, url_route: str, custom_llm_provider: str | None = None):
|
|
"""Check if the URL route is a Cursor Cloud Agents API route."""
|
|
if custom_llm_provider == "cursor":
|
|
return True
|
|
parsed_url: Final = urlparse(url_route)
|
|
if parsed_url.hostname and "api.cursor.com" in parsed_url.hostname:
|
|
return True
|
|
for route in self.TRACKED_CURSOR_ROUTES:
|
|
if route in url_route:
|
|
path = parsed_url.path if parsed_url.scheme else url_route
|
|
if path.startswith("/v0/"):
|
|
return custom_llm_provider == "cursor"
|
|
return False
|
|
|
|
def is_openai_route(self, url_route: str):
|
|
"""Check if the URL route is an OpenAI API route.
|
|
|
|
Uses the URL-aware helper so that non-OpenAI Azure Cognitive Services
|
|
(Speech, Vision, Language, ...) sharing the `*.cognitiveservices.azure.com`
|
|
/ `*.openai.azure.com` domains are not misclassified as OpenAI routes.
|
|
"""
|
|
if not url_route:
|
|
return False
|
|
from .llm_provider_handlers.openai_passthrough_logging_handler import (
|
|
_is_openai_compatible_url,
|
|
)
|
|
|
|
return _is_openai_compatible_url(url_route)
|
|
|
|
def is_gemini_route(self, url_route: str, custom_llm_provider: str | None = None):
|
|
"""Check if the URL route is a Gemini API route."""
|
|
if custom_llm_provider != "gemini":
|
|
return False
|
|
if VertexPassthroughLoggingHandler.is_interactions_route(url_route):
|
|
return True
|
|
for route in self.TRACKED_GEMINI_ROUTES:
|
|
if route in url_route:
|
|
return True
|
|
return False
|
|
|
|
def _is_supported_openai_endpoint(self, url_route: str) -> bool:
|
|
"""Check if the OpenAI endpoint is supported by the passthrough logging handler.
|
|
|
|
The Responses API route is included because
|
|
`openai_passthrough_handler` has a dedicated `elif is_responses:`
|
|
branch that knows how to extract usage + cost from the
|
|
Responses-API on-the-wire shape. Without including it here, the
|
|
outer dispatch filters Responses calls out before reaching the
|
|
handler — the inner branch is then unreachable and Responses
|
|
calls land in `LiteLLM_SpendLogs` with zero tokens / zero spend.
|
|
"""
|
|
from .llm_provider_handlers.openai_passthrough_logging_handler import (
|
|
OpenAIPassthroughLoggingHandler,
|
|
)
|
|
|
|
return (
|
|
OpenAIPassthroughLoggingHandler.is_openai_chat_completions_route(url_route)
|
|
or OpenAIPassthroughLoggingHandler.is_openai_embeddings_route(url_route)
|
|
or OpenAIPassthroughLoggingHandler.is_openai_image_generation_route(url_route)
|
|
or OpenAIPassthroughLoggingHandler.is_openai_image_editing_route(url_route)
|
|
or OpenAIPassthroughLoggingHandler.is_openai_responses_route(url_route)
|
|
)
|
|
|
|
def _set_cost_per_request(
|
|
self,
|
|
logging_obj: LiteLLMLoggingObj,
|
|
passthrough_logging_payload: PassthroughStandardLoggingPayload,
|
|
kwargs: dict,
|
|
):
|
|
"""
|
|
Helper function to set the cost per request in the logging object
|
|
|
|
Only set the cost per request if it's set in the passthrough logging payload.
|
|
If it's not set, don't set it in the logging object.
|
|
|
|
An upstream that prices its own requests always wins: ``cost_per_request``
|
|
is a flat per-request estimate for targets LiteLLM cannot price, and it
|
|
defaults to 0.0 on every config-defined endpoint, so honoring it here
|
|
would zero out the real cost the upstream reported. That holds even when
|
|
the reported value was unusable, where the contract records 0 rather
|
|
than billing an estimate the upstream just contradicted.
|
|
"""
|
|
#########################################################
|
|
# Check if cost per request is set
|
|
#########################################################
|
|
if has_upstream_reported_usage(logging_obj):
|
|
return kwargs
|
|
|
|
if passthrough_logging_payload.get("cost_per_request") is not None:
|
|
kwargs["response_cost"] = passthrough_logging_payload.get("cost_per_request")
|
|
logging_obj.model_call_details["response_cost"] = passthrough_logging_payload.get("cost_per_request")
|
|
|
|
return kwargs
|