diff --git a/litellm/constants.py b/litellm/constants.py index d1daf2ee2d1..e4fcb4909bd 100644 --- a/litellm/constants.py +++ b/litellm/constants.py @@ -422,6 +422,7 @@ WEBSOCKET_CLOSE_REASON_MAX_BYTES: Final = 123 DEEPGRAM_DEFAULT_API_BASE: Final = "https://api.deepgram.com/v1" NADIR_DEFAULT_API_BASE: Final = "https://api.getnadir.com/v1" DEEPGRAM_LISTEN_DEFAULT_MODEL: Final = "nova-3" +DEEPGRAM_SPEAK_DEFAULT_MODEL: Final = "aura-asteria-en" BEDROCK_REALTIME_PENDING_OPENING_MESSAGE_SCOPE_KEY: Final = "litellm.bedrock_realtime.pending_opening_message" BEDROCK_REALTIME_SESSION_COMMITTED_SCOPE_KEY: Final = "litellm.bedrock_realtime.session_committed" diff --git a/litellm/llms/deepgram/common_utils.py b/litellm/llms/deepgram/common_utils.py index 29998895bbc..a4a413370f7 100644 --- a/litellm/llms/deepgram/common_utils.py +++ b/litellm/llms/deepgram/common_utils.py @@ -1,16 +1,18 @@ import math +import re from collections.abc import Mapping, Sequence from types import MappingProxyType -from typing import Final +from typing import Final, Literal from urllib.parse import parse_qs, urlparse import httpx +from pydantic import BaseModel, ConfigDict, ValidationError import litellm -from litellm.constants import DEEPGRAM_DEFAULT_API_BASE, DEEPGRAM_LISTEN_DEFAULT_MODEL +from litellm.constants import DEEPGRAM_DEFAULT_API_BASE, DEEPGRAM_LISTEN_DEFAULT_MODEL, DEEPGRAM_SPEAK_DEFAULT_MODEL from litellm.litellm_core_utils.url_utils import join_api_base from litellm.llms.base_llm.chat.transformation import BaseLLMException -from litellm.types.utils import LlmProviders +from litellm.types.utils import LlmProviders, ModelInfo _WEBSOCKET_SCHEMES: Final = MappingProxyType({"https": "wss", "http": "ws", "wss": "wss", "ws": "ws"}) DEEPGRAM_LISTEN_CALLBACK_PARAMS: Final = frozenset({"callback", "callback_method"}) @@ -26,6 +28,11 @@ DEEPGRAM_LISTEN_ADDON_PRICING_PARAMS: Final = MappingProxyType( "diarize_model": "diarize", } ) +# Deepgram bills every voice of an Aura generation at one rate, so each generation has one cost-map row +DEEPGRAM_SPEAK_PRICING_FAMILIES: Final = ( + (re.compile(r"aura-2-[a-z]+-[a-z]{2}"), "aura-2"), + (re.compile(r"aura-[a-z]+-[a-z]{2}"), "aura"), +) _DISABLED_PARAM_VALUES: Final = frozenset({"", "false"}) _SINGLE_VALUED_PARAMS: Final = frozenset({"model", "language"}) @@ -34,8 +41,12 @@ class DeepgramException(BaseLLMException): pass +def _requested_model(query_string: str, default_model: str) -> str: + return httpx.QueryParams(query_string).get("model") or default_model + + def deepgram_listen_requested_model(query_string: str) -> str: - return httpx.QueryParams(query_string).get("model") or DEEPGRAM_LISTEN_DEFAULT_MODEL + return _requested_model(query_string, DEEPGRAM_LISTEN_DEFAULT_MODEL) def _first_occurrences(query_string: str) -> httpx.QueryParams: @@ -50,21 +61,29 @@ def _first_occurrences(query_string: str) -> httpx.QueryParams: ) -def deepgram_listen_websocket_target(api_base: str | None, query_string: str) -> str: - listen_url: Final = httpx.URL(join_api_base(api_base or DEEPGRAM_DEFAULT_API_BASE, "listen")) - websocket_url: Final = listen_url.copy_with(scheme=_WEBSOCKET_SCHEMES.get(listen_url.scheme, listen_url.scheme)) +def _deepgram_websocket_target(api_base: str | None, path: str, query_string: str, default_model: str) -> str: + http_url: Final = httpx.URL(join_api_base(api_base or DEEPGRAM_DEFAULT_API_BASE, path)) + websocket_url: Final = http_url.copy_with(scheme=_WEBSOCKET_SCHEMES.get(http_url.scheme, http_url.scheme)) params: Final = _first_occurrences(query_string) - query: Final = params if params.get("model") else params.remove("model").add("model", DEEPGRAM_LISTEN_DEFAULT_MODEL) + query: Final = params if params.get("model") else params.remove("model").add("model", default_model) return f"{websocket_url}?{query}" +def deepgram_listen_websocket_target(api_base: str | None, query_string: str) -> str: + return _deepgram_websocket_target(api_base, "listen", query_string, DEEPGRAM_LISTEN_DEFAULT_MODEL) + + def deepgram_listen_callback_params(query_string: str) -> tuple[str, ...]: return tuple(sorted(DEEPGRAM_LISTEN_CALLBACK_PARAMS.intersection(httpx.QueryParams(query_string).keys()))) -def deepgram_listen_model(upstream_url: str) -> str: +def _upstream_model(upstream_url: str, default_model: str) -> str: models: Final = parse_qs(urlparse(upstream_url).query).get("model") - return models[0] if models else DEEPGRAM_LISTEN_DEFAULT_MODEL + return models[0] if models else default_model + + +def deepgram_listen_model(upstream_url: str) -> str: + return _upstream_model(upstream_url, DEEPGRAM_LISTEN_DEFAULT_MODEL) def _param_enabled(values: Sequence[str]) -> bool: @@ -89,12 +108,15 @@ def deepgram_listen_registry_key(upstream_url: str) -> str: def deepgram_listen_is_priced(upstream_url: str) -> bool: """Only an exact registry hit counts: the cost calculator resolves a missing ``streaming/`` row to the pre-recorded ```` row, which is not the rate Deepgram bills a WebSocket session at.""" - registry_key: Final = deepgram_listen_registry_key(upstream_url) + return _exact_registry_entry(deepgram_listen_registry_key(upstream_url)) is not None + + +def _exact_registry_entry(registry_key: str) -> ModelInfo | None: try: model_info: Final = litellm.get_model_info(model=registry_key, custom_llm_provider=LlmProviders.DEEPGRAM.value) except Exception: - return False - return model_info["key"] == registry_key + return None + return model_info if model_info["key"] == registry_key else None def deepgram_listen_addon_pricing_models(upstream_url: str) -> tuple[str, ...]: @@ -117,10 +139,11 @@ def _channel_count(value: object) -> int | None: def _results_channel_count(frame: Mapping[str, object]) -> int | None: - channel_index: Final = frame.get("channel_index") - if not isinstance(channel_index, list) or len(channel_index) != 2: - return None - return _channel_count(channel_index[1]) + match frame.get("channel_index"): + case [object(), int() as channels]: + return _channel_count(channels) + case _: + return None def _declared_channel_count(upstream_url: str) -> int | None: @@ -199,3 +222,50 @@ def deepgram_listen_transcript(websocket_messages: Sequence[Mapping[str, object] if frame.get("type") == "Results" if (transcript := _final_transcript(frame)) is not None ) + + +def deepgram_speak_requested_model(query_string: str) -> str: + return _requested_model(query_string, DEEPGRAM_SPEAK_DEFAULT_MODEL) + + +def deepgram_speak_websocket_target(api_base: str | None, query_string: str) -> str: + return _deepgram_websocket_target(api_base, "speak", query_string, DEEPGRAM_SPEAK_DEFAULT_MODEL) + + +def deepgram_speak_model(upstream_url: str) -> str: + return _upstream_model(upstream_url, DEEPGRAM_SPEAK_DEFAULT_MODEL) + + +def deepgram_speak_registry_key(upstream_url: str) -> str: + """The row a speak session is billed at: its Aura generation's row, else a row named after the voice itself""" + model: Final = deepgram_speak_model(upstream_url) + pricing_model: Final = next( + (family for pattern, family in DEEPGRAM_SPEAK_PRICING_FAMILIES if pattern.fullmatch(model)), model + ) + return f"{LlmProviders.DEEPGRAM.value}/{pricing_model}" + + +def deepgram_speak_character_rate(upstream_url: str) -> float | None: + model_info: Final = _exact_registry_entry(deepgram_speak_registry_key(upstream_url)) + return None if model_info is None else model_info.get("input_cost_per_character") + + +def deepgram_speak_is_priced(upstream_url: str) -> bool: + return deepgram_speak_character_rate(upstream_url) is not None + + +class _DeepgramSpeakMessage(BaseModel): + model_config = ConfigDict(frozen=True) + + type: Literal["Speak"] + text: str + + +def deepgram_speak_characters(frame: str | bytes) -> int: + """Characters Deepgram bills for one client frame: the length of a Speak message's text, else 0. It never raises, + because the relay runs it on every frame and a traceback from a malformed frame would carry text that must not be + logged""" + try: + return len(_DeepgramSpeakMessage.model_validate_json(frame).text) + except ValidationError: + return 0 diff --git a/litellm/model_prices_and_context_window_backup.json b/litellm/model_prices_and_context_window_backup.json index b820b058eb3..41070e908ee 100644 --- a/litellm/model_prices_and_context_window_backup.json +++ b/litellm/model_prices_and_context_window_backup.json @@ -22082,6 +22082,32 @@ "output_cost_per_token_batches": 1e-06, "source": "https://developers.openai.com/api/docs/pricing" }, + "deepgram/aura": { + "input_cost_per_character": 1.5e-05, + "litellm_provider": "deepgram", + "metadata": { + "calculation": "$0.0150/1000 characters = $0.000015 per character", + "note": "Aura-1 text-to-speech voices (aura--), pay as you go" + }, + "mode": "audio_speech", + "source": "https://deepgram.com/pricing", + "supported_endpoints": [ + "/v1/speak" + ] + }, + "deepgram/aura-2": { + "input_cost_per_character": 3e-05, + "litellm_provider": "deepgram", + "metadata": { + "calculation": "$0.030/1000 characters = $0.00003 per character", + "note": "Aura-2 text-to-speech voices (aura-2--), pay as you go" + }, + "mode": "audio_speech", + "source": "https://deepgram.com/pricing", + "supported_endpoints": [ + "/v1/speak" + ] + }, "deepgram/base": { "input_cost_per_second": 0.00020833, "litellm_provider": "deepgram", diff --git a/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py b/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py index 63b3c300e7f..e29b27c4fd2 100644 --- a/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py @@ -63,6 +63,11 @@ from litellm.llms.deepgram.common_utils import ( deepgram_listen_registry_key, deepgram_listen_requested_model, deepgram_listen_websocket_target, + deepgram_speak_characters, + deepgram_speak_is_priced, + deepgram_speak_registry_key, + deepgram_speak_requested_model, + deepgram_speak_websocket_target, ) from litellm.llms.fal_ai.cost_calculator import fal_ai_passthrough_cost, fal_ai_queue_base from litellm.llms.nvidia_nim.passthrough.transformation import nvidia_nim_model_group_in_path @@ -3341,6 +3346,7 @@ class _WebsocketRelay(Protocol): forward_headers: bool, endpoint: str, accept_websocket: bool, + count_client_text: Callable[[str | bytes], int] | None = None, ) -> None: ... @@ -3461,12 +3467,22 @@ async def deepgram_listen_user_api_key_auth(websocket: WebSocket) -> UserAPIKeyA ) -@router.websocket("/deepgram/v1/listen") -@router.websocket("/deepgram/listen") -async def deepgram_listen_websocket_route( +def _deepgram_listen_refusal(target: str) -> str | None: + callback_params: Final = deepgram_listen_callback_params(httpx.URL(target).query.decode()) + if callback_params: + return _DEEPGRAM_WS_CALLBACK_REASON.format(params=", ".join(callback_params)) + if deepgram_listen_is_priced(target): + return None + return _DEEPGRAM_WS_UNPRICED_REASON.format(registry_key=deepgram_listen_registry_key(target)) + + +async def _relay_deepgram_websocket( websocket: WebSocket, - user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(deepgram_listen_user_api_key_auth)], - relay: Annotated[_WebsocketRelay, Depends(_websocket_relay)], + user_api_key_dict: UserAPIKeyAuth, + relay: _WebsocketRelay, + target: str, + refusal: str | None, + count_client_text: Callable[[str | bytes], int] | None = None, ) -> None: deepgram_api_key: Final = passthrough_endpoint_router.get_credentials( custom_llm_provider=litellm.LlmProviders.DEEPGRAM.value, @@ -3477,23 +3493,8 @@ async def deepgram_listen_websocket_route( return await websocket.accept(subprotocol=_negotiated_websocket_subprotocol(websocket)) - callback_params: Final = deepgram_listen_callback_params(websocket.url.query) - if callback_params: - await websocket.close( - code=1008, - reason=_DEEPGRAM_WS_CALLBACK_REASON.format(params=", ".join(callback_params)), - ) - return - - target: Final = deepgram_listen_websocket_target( - api_base=get_secret_str("DEEPGRAM_API_BASE"), - query_string=websocket.url.query, - ) - if not deepgram_listen_is_priced(target): - await websocket.close( - code=1008, - reason=_DEEPGRAM_WS_UNPRICED_REASON.format(registry_key=deepgram_listen_registry_key(target)), - ) + if refusal is not None: + await websocket.close(code=1008, reason=refusal) return await relay( @@ -3504,6 +3505,50 @@ async def deepgram_listen_websocket_route( forward_headers=False, endpoint=websocket.url.path, accept_websocket=False, + count_client_text=count_client_text, + ) + + +@router.websocket("/deepgram/v1/listen") +@router.websocket("/deepgram/listen") +async def deepgram_listen_websocket_route( + websocket: WebSocket, + user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(deepgram_listen_user_api_key_auth)], + relay: Annotated[_WebsocketRelay, Depends(_websocket_relay)], +) -> None: + target: Final = deepgram_listen_websocket_target(get_secret_str("DEEPGRAM_API_BASE"), websocket.url.query) + await _relay_deepgram_websocket(websocket, user_api_key_dict, relay, target, _deepgram_listen_refusal(target)) + + +async def deepgram_speak_user_api_key_auth(websocket: WebSocket) -> UserAPIKeyAuth: + return await user_api_key_auth_websocket_for_model( + websocket, + model=deepgram_speak_requested_model(websocket.url.query), + accepted_authorization_schemes=_DEEPGRAM_WS_AUTHORIZATION_SCHEMES, + ) + + +def _deepgram_speak_refusal(target: str) -> str | None: + if deepgram_speak_is_priced(target): + return None + return _DEEPGRAM_WS_UNPRICED_REASON.format(registry_key=deepgram_speak_registry_key(target)) + + +@router.websocket("/deepgram/v1/speak") +@router.websocket("/deepgram/speak") +async def deepgram_speak_websocket_route( + websocket: WebSocket, + user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(deepgram_speak_user_api_key_auth)], + relay: Annotated[_WebsocketRelay, Depends(_websocket_relay)], +) -> None: + target: Final = deepgram_speak_websocket_target(get_secret_str("DEEPGRAM_API_BASE"), websocket.url.query) + await _relay_deepgram_websocket( + websocket, + user_api_key_dict, + relay, + target, + _deepgram_speak_refusal(target), + count_client_text=deepgram_speak_characters, ) diff --git a/litellm/proxy/pass_through_endpoints/llm_provider_handlers/deepgram_speak_passthrough_logging_handler.py b/litellm/proxy/pass_through_endpoints/llm_provider_handlers/deepgram_speak_passthrough_logging_handler.py new file mode 100644 index 00000000000..572d183ecc2 --- /dev/null +++ b/litellm/proxy/pass_through_endpoints/llm_provider_handlers/deepgram_speak_passthrough_logging_handler.py @@ -0,0 +1,79 @@ +import json +from collections.abc import Mapping +from types import MappingProxyType +from typing import Final +from urllib.parse import urlparse + +import litellm +from litellm._logging import verbose_proxy_logger +from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj +from litellm.llms.deepgram.common_utils import ( + deepgram_speak_character_rate, + deepgram_speak_model, + deepgram_speak_registry_key, +) +from litellm.proxy._types import PassThroughEndpointLoggingTypedDict +from litellm.types.passthrough_endpoints.pass_through_endpoints import WEBSOCKET_CLIENT_TEXT_COUNT_KEY +from litellm.types.utils import StandardPassThroughResponseObject + +DEEPGRAM_SPEAK_ROUTE_SUFFIX: Final = "/speak" + + +def _billed_characters(request_body: Mapping[str, object]) -> int: + characters: Final = request_body.get(WEBSOCKET_CLIENT_TEXT_COUNT_KEY) + return characters if isinstance(characters, int) else 0 + + +def _speech_cost(upstream_url: str, characters: int) -> float | None: + rate: Final = deepgram_speak_character_rate(upstream_url) + if rate is None: + verbose_proxy_logger.warning( + "Deepgram speak passthrough: no per-character registry entry '%s'", + deepgram_speak_registry_key(upstream_url), + ) + return None + return characters * rate + + +class DeepgramSpeakPassthroughLoggingHandler: + """Bills a Deepgram speak session by the characters of Speak text the client sent. The relay hands over only the + count, never the text, so the spend row and every callback see a model, a count and a cost and nothing the caller + asked to have spoken""" + + @staticmethod + def is_deepgram_speak_route(url_route: str, request_method: str) -> bool: + path: Final = urlparse(url_route).path + return request_method == "WEBSOCKET" and "/deepgram/" in path and path.endswith(DEEPGRAM_SPEAK_ROUTE_SUFFIX) + + @staticmethod + def deepgram_speak_passthrough_handler( + request_body: Mapping[str, object], + logging_obj: LiteLLMLoggingObj, + upstream_url: str, + kwargs: Mapping[str, object] = MappingProxyType({}), + ) -> PassThroughEndpointLoggingTypedDict: + model: Final = deepgram_speak_model(upstream_url) + characters: Final = _billed_characters(request_body) + response_cost: Final = _speech_cost(upstream_url, characters) + + provider: Final = litellm.LlmProviders.DEEPGRAM.value + logging_obj.model = model # rebind-ok: the spend logger reads model and cost off the shared logging object + logging_obj.model_call_details["model"] = model # rebind-ok: same shared logging object + logging_obj.model_call_details["custom_llm_provider"] = provider # rebind-ok: same shared logging object + logging_obj.model_call_details["response_cost"] = response_cost # rebind-ok: same shared logging object + verbose_proxy_logger.debug( + "Deepgram speak passthrough cost tracking: model %s, characters %s, cost %s", + model, + characters, + response_cost, + ) + logging_result: Final[PassThroughEndpointLoggingTypedDict] = { + "result": StandardPassThroughResponseObject(response=json.dumps({"characters": characters})), + "kwargs": { + **kwargs, + "model": model, + "custom_llm_provider": provider, + "response_cost": response_cost, + }, + } + return logging_result diff --git a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py index f8953c44ba2..a19ac2e66ef 100644 --- a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py @@ -137,6 +137,7 @@ from litellm.types.passthrough_endpoints.pass_through_endpoints import ( LITELLM_PASS_THROUGH_DEPLOYMENT_MODEL_INFO_STATE_KEY, LITELLM_PASS_THROUGH_ENDPOINT_MARKER, LITELLM_PASS_THROUGH_RAW_BODY_STATE_KEY, + WEBSOCKET_CLIENT_TEXT_COUNT_KEY, EndpointType, PassthroughStandardLoggingPayload, ) @@ -2476,13 +2477,14 @@ def _with_trace_context(headers: Mapping[str, str], parent_span: object) -> dict async def websocket_passthrough_request( websocket: WebSocket, target: str, - custom_headers: dict, + custom_headers: Mapping[str, str], user_api_key_dict: UserAPIKeyAuth, forward_headers: bool | None = False, endpoint: str | None = None, cost_per_request: float | None = None, accept_websocket: bool = True, setup_model_rewriter: Callable[[str], str] | None = None, + count_client_text: Callable[[str | bytes], int] | None = None, ): """ WebSocket passthrough request handler. @@ -2496,6 +2498,9 @@ async def websocket_passthrough_request( endpoint: The endpoint path (for logging purposes) cost_per_request: Optional field - cost per request to the target endpoint setup_model_rewriter: Optional rewrite of the setup frame's model before it reaches the upstream + count_client_text: Optional read-only count of each client frame, text or binary, after it reaches the + upstream, such as billable characters. Only the sum reaches the success handler, under WEBSOCKET_CLIENT_TEXT_COUNT_KEY + in request_body, so routes whose client text must not be stored can still bill it """ from litellm.litellm_core_utils.litellm_logging import Logging from litellm.proxy.proxy_server import proxy_config, proxy_logging_obj @@ -2506,6 +2511,7 @@ async def websocket_passthrough_request( # Initialize tracking variables start_time: Final = datetime.now() websocket_messages: Final[list[dict[str, object]]] = [] + client_text_count: int = 0 # rebind-ok: forward_client_to_upstream adds each client frame's count litellm_call_id: Final = str(uuid.uuid4()) verbose_proxy_logger.info("WebSocket passthrough (%s): Starting WebSocket connection to %s", endpoint, target) @@ -2622,6 +2628,16 @@ async def websocket_passthrough_request( "WebSocket passthrough (%s): Upstream connection established successfully", endpoint ) + async def send_upstream(frame: str | bytes) -> None: + nonlocal client_text_count + billed: Final = 0 if count_client_text is None else count_client_text(frame) + try: + await upstream_ws.send(frame) + except asyncio.CancelledError: + client_text_count += billed # rebind-ok: the frame can already be on the wire when the cancel lands + raise + client_text_count += billed # rebind-ok: the session total grows with each delivered frame + async def forward_client_to_upstream() -> None: """Forward messages from client to upstream WebSocket""" try: @@ -2692,9 +2708,9 @@ async def websocket_passthrough_request( ) # Not a JSON message or doesn't contain setup data - await upstream_ws.send(_rewrite_vertex_live_setup_model(text_data, setup_model_rewriter)) + await send_upstream(_rewrite_vertex_live_setup_model(text_data, setup_model_rewriter)) elif bytes_data is not None: - await upstream_ws.send(bytes_data) + await send_upstream(bytes_data) except asyncio.CancelledError: raise except Exception: @@ -2835,7 +2851,9 @@ async def websocket_passthrough_request( end_time=end_time, logging_obj=logging_obj, cache_hit=False, - request_body={}, + request_body={} + if count_client_text is None + else {WEBSOCKET_CLIENT_TEXT_COUNT_KEY: client_text_count}, **success_kwargs, ) ) diff --git a/litellm/proxy/pass_through_endpoints/success_handler.py b/litellm/proxy/pass_through_endpoints/success_handler.py index 9c6c3111209..8ad1a221f52 100644 --- a/litellm/proxy/pass_through_endpoints/success_handler.py +++ b/litellm/proxy/pass_through_endpoints/success_handler.py @@ -29,6 +29,9 @@ from .llm_provider_handlers.cursor_passthrough_logging_handler import ( from .llm_provider_handlers.deepgram_listen_passthrough_logging_handler import ( DeepgramListenPassthroughLoggingHandler, ) +from .llm_provider_handlers.deepgram_speak_passthrough_logging_handler import ( + DeepgramSpeakPassthroughLoggingHandler, +) from .llm_provider_handlers.fal_ai_passthrough_logging_handler import ( FalAIPassthroughLoggingHandler, ) @@ -150,7 +153,7 @@ class PassThroughEndpointLogging: _handle_logging = handle_logging - def normalize_llm_passthrough_logging_payload( + def normalize_llm_passthrough_logging_payload( # noqa: C901 # provider dispatch, one branch per provider self, httpx_response: httpx.Response, response_body: dict | list[dict[str, object]] | None, @@ -400,6 +403,17 @@ class PassThroughEndpointLogging: ) standard_logging_response_object = deepgram_handler_result["result"] # rebind-ok: elif-chain kwargs = deepgram_handler_result["kwargs"] # rebind-ok: elif-chain contract + elif DeepgramSpeakPassthroughLoggingHandler.is_deepgram_speak_route(url_route, httpx_response.request.method): + deepgram_speak_handler_result: Final = ( + DeepgramSpeakPassthroughLoggingHandler.deepgram_speak_passthrough_handler( + request_body=request_body, + logging_obj=logging_obj, + upstream_url=str(httpx_response.request.url), + kwargs=kwargs, + ) + ) + standard_logging_response_object = deepgram_speak_handler_result["result"] # rebind-ok: elif-chain + kwargs = deepgram_speak_handler_result["kwargs"] # rebind-ok: elif-chain contract elif FalAIPassthroughLoggingHandler.is_fal_ai_route(url_route, custom_llm_provider): fal_ai_handler_result: Final = FalAIPassthroughLoggingHandler().fal_ai_passthrough_handler( response_body=response_body if isinstance(response_body, dict) else MappingProxyType({}), diff --git a/litellm/types/passthrough_endpoints/pass_through_endpoints.py b/litellm/types/passthrough_endpoints/pass_through_endpoints.py index 0d99cd54a6b..40a78d79194 100644 --- a/litellm/types/passthrough_endpoints/pass_through_endpoints.py +++ b/litellm/types/passthrough_endpoints/pass_through_endpoints.py @@ -27,6 +27,8 @@ LITELLM_PASS_THROUGH_ENDPOINT_MARKER: Final = "__litellm_pass_through_endpoint__ SETTLES_BACKGROUND_RESPONSE_KWARG: Final = "settles_background_response" +WEBSOCKET_CLIENT_TEXT_COUNT_KEY: Final = "litellm_websocket_client_text_count" + class BackgroundResponseSettlement(str, Enum): SETTLES = "settles" diff --git a/model_prices_and_context_window.json b/model_prices_and_context_window.json index b820b058eb3..41070e908ee 100644 --- a/model_prices_and_context_window.json +++ b/model_prices_and_context_window.json @@ -22082,6 +22082,32 @@ "output_cost_per_token_batches": 1e-06, "source": "https://developers.openai.com/api/docs/pricing" }, + "deepgram/aura": { + "input_cost_per_character": 1.5e-05, + "litellm_provider": "deepgram", + "metadata": { + "calculation": "$0.0150/1000 characters = $0.000015 per character", + "note": "Aura-1 text-to-speech voices (aura--), pay as you go" + }, + "mode": "audio_speech", + "source": "https://deepgram.com/pricing", + "supported_endpoints": [ + "/v1/speak" + ] + }, + "deepgram/aura-2": { + "input_cost_per_character": 3e-05, + "litellm_provider": "deepgram", + "metadata": { + "calculation": "$0.030/1000 characters = $0.00003 per character", + "note": "Aura-2 text-to-speech voices (aura-2--), pay as you go" + }, + "mode": "audio_speech", + "source": "https://deepgram.com/pricing", + "supported_endpoints": [ + "/v1/speak" + ] + }, "deepgram/base": { "input_cost_per_second": 0.00020833, "litellm_provider": "deepgram", diff --git a/tests/unit/llms/deepgram/test_deepgram_common_utils.py b/tests/unit/llms/deepgram/test_deepgram_common_utils.py index 530888b70c4..e5a95400a76 100644 --- a/tests/unit/llms/deepgram/test_deepgram_common_utils.py +++ b/tests/unit/llms/deepgram/test_deepgram_common_utils.py @@ -1,3 +1,4 @@ +import json import math from collections.abc import Mapping, Sequence from typing import Final @@ -18,6 +19,13 @@ from litellm.llms.deepgram.common_utils import ( deepgram_listen_requested_model, deepgram_listen_transcript, deepgram_listen_websocket_target, + deepgram_speak_character_rate, + deepgram_speak_characters, + deepgram_speak_is_priced, + deepgram_speak_model, + deepgram_speak_registry_key, + deepgram_speak_requested_model, + deepgram_speak_websocket_target, ) NOVA_3_URL: Final = "wss://api.deepgram.com/v1/listen?model=nova-3&encoding=linear16&sample_rate=16000" @@ -324,3 +332,163 @@ def test_deepgram_listen_is_priced( ) def test_deepgram_listen_addon_pricing_models(upstream_url: str, expected: tuple[str, ...]): assert deepgram_listen_addon_pricing_models(upstream_url) == expected + + +@pytest.mark.usefixtures("local_model_cost_map") +@pytest.mark.parametrize( + ("pricing_model", "usd_per_thousand_characters"), + [ + pytest.param("aura-2", 0.030, id="aura-2"), + pytest.param("aura", 0.015, id="aura-1"), + ], +) +def test_aura_rows_bill_speech_per_character_at_the_published_rate( + pricing_model: str, usd_per_thousand_characters: float +) -> None: + """Pay as you go rates from https://deepgram.com/pricing, checked 2026-10-26""" + prompt_cost, completion_cost = litellm.cost_per_token( + model=pricing_model, custom_llm_provider="deepgram", call_type="speech", prompt_characters=1000 + ) + + assert prompt_cost == pytest.approx(usd_per_thousand_characters) + assert completion_cost == 0.0 + + +SPEAK_BASE: Final = "wss://api.deepgram.com/v1/speak" + + +@pytest.mark.parametrize( + ("api_base", "query_string", "expected"), + [ + pytest.param(None, "", f"{SPEAK_BASE}?model=aura-asteria-en", id="default model added"), + pytest.param(None, "model=", f"{SPEAK_BASE}?model=aura-asteria-en", id="empty model replaced"), + pytest.param( + None, + "model=aura-2-thalia-en&encoding=linear16&sample_rate=24000", + f"{SPEAK_BASE}?model=aura-2-thalia-en&encoding=linear16&sample_rate=24000", + id="query kept in order", + ), + pytest.param( + None, + "model=aura-2-thalia-en&model=aura-2-zeus-en", + f"{SPEAK_BASE}?model=aura-2-thalia-en", + id="only the first model", + ), + pytest.param( + "http://127.0.0.1:8080/v1", "model=aura-2-thalia-en", "ws://127.0.0.1:8080/v1/speak?model=aura-2-thalia-en" + ), + pytest.param( + "https://dg.internal/v1/", + "encoding=mulaw", + "wss://dg.internal/v1/speak?encoding=mulaw&model=aura-asteria-en", + ), + ], +) +def test_deepgram_speak_websocket_target(api_base: str | None, query_string: str, expected: str) -> None: + assert deepgram_speak_websocket_target(api_base=api_base, query_string=query_string) == expected + + +@pytest.mark.parametrize( + "query_string", + ["", "model=aura-2-thalia-en", "model=aura-2-thalia-en&model=aura-2-zeus-en", "encoding=mulaw&model=aura-luna-en"], +) +def test_the_authorized_speak_model_is_the_only_model_the_upstream_target_carries(query_string: str) -> None: + """The key is checked against deepgram_speak_requested_model, so the upstream must use exactly that voice""" + target: Final = deepgram_speak_websocket_target(api_base=None, query_string=query_string) + + assert parse_qs(urlparse(target).query)["model"] == [deepgram_speak_requested_model(query_string)] + assert deepgram_speak_model(target) == deepgram_speak_requested_model(query_string) + + +@pytest.mark.parametrize( + ("model", "expected_family"), + [ + # Voice names from https://developers.deepgram.com/docs/tts-models, checked 2026-10-26 + pytest.param("aura-2-thalia-en", "aura-2", id="aura-2 english"), + pytest.param("aura-2-celeste-es", "aura-2", id="aura-2 spanish"), + pytest.param("aura-2-rhea-nl", "aura-2", id="aura-2 dutch"), + pytest.param("aura-2-agathe-fr", "aura-2", id="aura-2 french"), + pytest.param("aura-2-viktoria-de", "aura-2", id="aura-2 german"), + pytest.param("aura-2-livia-it", "aura-2", id="aura-2 italian"), + pytest.param("aura-2-izanami-ja", "aura-2", id="aura-2 japanese"), + pytest.param("aura-asteria-en", "aura", id="aura-1"), + pytest.param("aura-zeus-en", "aura", id="another aura-1"), + pytest.param("aura-3-foo-en", "aura-3-foo-en", id="unknown aura generation"), + pytest.param("aura-2", "aura-2", id="bare aura-2"), + pytest.param("aura-x", "aura-x", id="aura without a language"), + pytest.param("aura-2-thalia-en-extra", "aura-2-thalia-en-extra", id="trailing part"), + pytest.param("flux-tts-en", "flux-tts-en", id="other family"), + pytest.param("nova-3", "nova-3", id="listen model"), + ], +) +def test_deepgram_speak_registry_key_names_the_aura_generation_row_else_the_voice( + model: str, expected_family: str +) -> None: + target: Final = deepgram_speak_websocket_target(api_base=None, query_string=f"model={model}") + + assert deepgram_speak_registry_key(target) == f"deepgram/{expected_family}" + + +@pytest.mark.usefixtures("local_model_cost_map") +@pytest.mark.parametrize( + ("model", "expected"), + [ + pytest.param("aura-2-thalia-en", True, id="aura-2"), + pytest.param("aura-asteria-en", True, id="aura-1"), + pytest.param("aura-3-foo-en", False, id="unknown aura generation"), + pytest.param("flux-tts-en", False, id="unpriced family"), + pytest.param("nova-3", False, id="a listen model has no speech row"), + ], +) +def test_deepgram_speak_is_priced(model: str, expected: bool) -> None: + assert deepgram_speak_is_priced(f"{SPEAK_BASE}?model={model}") is expected + + +@pytest.mark.usefixtures("local_model_cost_map") +def test_deepgram_speak_is_not_priced_once_its_family_row_is_gone(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delitem(litellm.model_cost, "deepgram/aura-2") + + assert deepgram_speak_is_priced(f"{SPEAK_BASE}?model=aura-2-thalia-en") is False + + +@pytest.mark.usefixtures("local_model_cost_map") +def test_a_cost_map_row_named_after_a_new_voice_prices_it_at_that_rate(monkeypatch: pytest.MonkeyPatch) -> None: + """The 1008 refusal tells the operator to add the voice's row, so adding it must be enough""" + url: Final = f"{SPEAK_BASE}?model=aura-3-foo-en" + monkeypatch.setitem( + litellm.model_cost, + "deepgram/aura-3-foo-en", + {"input_cost_per_character": 4e-05, "litellm_provider": "deepgram", "mode": "audio_speech"}, + ) + + assert deepgram_speak_character_rate(url) == 4e-05 + assert deepgram_speak_is_priced(url) is True + + +@pytest.mark.parametrize( + ("frame", "expected"), + [ + pytest.param( + json.dumps({"type": "Speak", "text": "The quick brown fox jumps over the lazy dog."}), 44, id="speak" + ), + pytest.param(json.dumps({"type": "Speak", "text": "Gr\u00fc\u00dfe \U0001f44b"}), 7, id="code points"), + pytest.param( + json.dumps({"type": "Speak", "text": "Sent as a binary frame."}).encode(), 23, id="speak in a binary frame" + ), + pytest.param(b"\xff\xfe\x00", 0, id="binary frame that is not utf-8"), + pytest.param(json.dumps({"type": "Speak", "text": ""}), 0, id="empty text"), + pytest.param(json.dumps({"type": "Flush"}), 0, id="flush"), + pytest.param(json.dumps({"type": "Clear"}), 0, id="clear"), + pytest.param(json.dumps({"type": "Close"}), 0, id="close"), + pytest.param(json.dumps({"type": "Speak"}), 0, id="speak without text"), + pytest.param(json.dumps({"type": "Speak", "text": 12345}), 0, id="text not a string"), + pytest.param(json.dumps({"type": "Speak", "text": ["a", "b"]}), 0, id="text a list"), + pytest.param(json.dumps([{"type": "Speak", "text": "hello"}]), 0, id="not an object"), + pytest.param("hello there", 0, id="not json"), + pytest.param("", 0, id="empty frame"), + pytest.param("[" * 100_000, 0, id="nesting deep enough to recurse"), + pytest.param("1" * 5_000, 0, id="integer longer than the parser allows"), + ], +) +def test_deepgram_speak_characters_counts_speak_text_and_never_raises(frame: str | bytes, expected: int) -> None: + assert deepgram_speak_characters(frame) == expected diff --git a/tests/unit/proxy/pass_through_endpoints/llm_provider_handlers/test_deepgram_speak_passthrough_logging_handler.py b/tests/unit/proxy/pass_through_endpoints/llm_provider_handlers/test_deepgram_speak_passthrough_logging_handler.py new file mode 100644 index 00000000000..82e0d388607 --- /dev/null +++ b/tests/unit/proxy/pass_through_endpoints/llm_provider_handlers/test_deepgram_speak_passthrough_logging_handler.py @@ -0,0 +1,174 @@ +"""Deepgram ``/v1/speak`` WebSocket passthrough: per-character cost tracking from the relay's client text count.""" + +import json +from datetime import datetime +from types import SimpleNamespace +from typing import Final + +import pytest + +import litellm +from litellm.integrations.custom_logger import CustomLogger +from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj +from litellm.proxy.pass_through_endpoints.llm_provider_handlers.deepgram_speak_passthrough_logging_handler import ( + DeepgramSpeakPassthroughLoggingHandler, +) +from litellm.proxy.pass_through_endpoints.success_handler import PassThroughEndpointLogging +from litellm.types.passthrough_endpoints.pass_through_endpoints import ( + WEBSOCKET_CLIENT_TEXT_COUNT_KEY, + PassthroughStandardLoggingPayload, +) +from litellm.types.utils import StandardLoggingPayload + +AURA_2_URL: Final = "wss://api.deepgram.com/v1/speak?model=aura-2-thalia-en&encoding=linear16&sample_rate=24000" +AURA_1_URL: Final = "wss://api.deepgram.com/v1/speak?model=aura-asteria-en" +SPOKEN_CHARACTERS: Final = len("The quick brown fox jumps over the lazy dog.") +# Pay as you go rates from https://deepgram.com/pricing, checked 2026-10-26 +AURA_2_USD_PER_CHARACTER: Final = 0.030 / 1000 +AURA_1_USD_PER_CHARACTER: Final = 0.0150 / 1000 + +pytestmark: Final = pytest.mark.usefixtures("local_model_cost_map") + + +@pytest.mark.parametrize( + ("url_route", "request_method", "expected"), + [ + ("/deepgram/v1/speak", "WEBSOCKET", True), + ("/deepgram/speak", "WEBSOCKET", True), + ("/deepgram/v1/speak?model=aura-2-thalia-en", "WEBSOCKET", True), + ("/deepgram/v1/listen", "WEBSOCKET", False), + ("/deepgram/v1/speak/extra", "WEBSOCKET", False), + ("/openai/v1/audio/speech", "WEBSOCKET", False), + ("", "WEBSOCKET", False), + pytest.param("https://gateway.internal/deepgram/v1/speak", "POST", False, id="http passthrough to a speak url"), + ], +) +def test_is_deepgram_speak_route(url_route: str, request_method: str, expected: bool) -> None: + assert DeepgramSpeakPassthroughLoggingHandler.is_deepgram_speak_route(url_route, request_method) is expected + + +def _logging_obj(call_id: str = "call-dg-speak") -> LiteLLMLoggingObj: + return LiteLLMLoggingObj( + model="unknown", + messages=[{"role": "user", "content": "WebSocket connection"}], + stream=True, + call_type="pass_through_endpoint", + start_time=datetime.now(), + litellm_call_id=call_id, + function_id="websocket_passthrough", + ) + + +def _handler_cost(upstream_url: str, characters: object) -> float | None: + handler_result: Final = DeepgramSpeakPassthroughLoggingHandler.deepgram_speak_passthrough_handler( + request_body={WEBSOCKET_CLIENT_TEXT_COUNT_KEY: characters}, + logging_obj=_logging_obj(), + upstream_url=upstream_url, + ) + response_cost: Final = handler_result["kwargs"]["response_cost"] + assert response_cost is None or isinstance(response_cost, float) + return response_cost + + +def test_handler_bills_spoken_characters_at_the_aura_2_rate_and_names_the_voice() -> None: + logging_obj: Final = _logging_obj() + + handler_result: Final = DeepgramSpeakPassthroughLoggingHandler.deepgram_speak_passthrough_handler( + request_body={WEBSOCKET_CLIENT_TEXT_COUNT_KEY: SPOKEN_CHARACTERS}, + logging_obj=logging_obj, + upstream_url=AURA_2_URL, + kwargs={"litellm_params": {"metadata": {}}}, + ) + + assert handler_result["kwargs"]["response_cost"] == pytest.approx(SPOKEN_CHARACTERS * AURA_2_USD_PER_CHARACTER) + assert handler_result["kwargs"]["model"] == "aura-2-thalia-en" + assert handler_result["kwargs"]["custom_llm_provider"] == "deepgram" + assert handler_result["kwargs"]["litellm_params"] == {"metadata": {}} + assert handler_result["result"] == {"response": json.dumps({"characters": SPOKEN_CHARACTERS})} + assert logging_obj.model == "aura-2-thalia-en" + assert logging_obj.model_call_details["response_cost"] == pytest.approx( + SPOKEN_CHARACTERS * AURA_2_USD_PER_CHARACTER + ) + + +def test_handler_bills_an_aura_1_voice_at_the_aura_1_rate() -> None: + assert _handler_cost(AURA_1_URL, 1000) == pytest.approx(1000 * AURA_1_USD_PER_CHARACTER) + + +@pytest.mark.parametrize( + "characters", + [pytest.param(0, id="nothing spoken"), pytest.param(None, id="no count")], +) +def test_handler_bills_zero_for_a_session_without_speak_text(characters: object) -> None: + assert _handler_cost(AURA_2_URL, characters) == 0.0 + + +@pytest.mark.parametrize("model", ["flux-tts-en", "aura-3-x-en", "aura-x", "nova-3"]) +def test_handler_reports_no_cost_rather_than_a_wrong_one_for_an_unpriced_voice(model: str) -> None: + assert _handler_cost(f"wss://api.deepgram.com/v1/speak?model={model}", 1000) is None + + +class _CapturingLogger(CustomLogger): + def __init__(self) -> None: + super().__init__() + self.payloads: list[StandardLoggingPayload] = [] # mutable-ok: the logger appends each payload it receives + + async def async_log_success_event(self, kwargs, response_obj, start_time, end_time) -> None: + self.payloads.append(kwargs["standard_logging_object"]) + + +@pytest.mark.parametrize( + ("characters", "expected_cost"), + [ + pytest.param(SPOKEN_CHARACTERS, SPOKEN_CHARACTERS * AURA_2_USD_PER_CHARACTER, id="one speak"), + pytest.param(0, 0.0, id="zero stays zero and is not recomputed"), + ], +) +async def test_success_handler_dispatches_deepgram_speak_and_logs_character_based_spend( + monkeypatch: pytest.MonkeyPatch, characters: int, expected_cost: float +) -> None: + """Drives the shared passthrough success handler the way the relay does at socket close and reads what a spend + logger receives""" + capturing_logger: Final = _CapturingLogger() + monkeypatch.setattr(litellm, "_async_success_callback", [capturing_logger]) + monkeypatch.setattr(litellm, "success_callback", []) + monkeypatch.setattr(litellm, "callbacks", []) + logging_obj: Final = _logging_obj("call-dg-speak-e2e") + litellm_params: Final = {"metadata": {"user_api_key": "hashed-key", "user_api_key_team_id": "team-tts"}} + logging_obj.update_environment_variables( + model="unknown", + user="unknown", + optional_params={}, + litellm_params=litellm_params, + call_type="pass_through_endpoint", + ) + + await PassThroughEndpointLogging().pass_through_async_success_handler( + httpx_response=SimpleNamespace( + status_code=200, + text="WebSocket connection successful", + headers={}, + request=SimpleNamespace(method="WEBSOCKET", url=AURA_2_URL), + ), + response_body=[{"type": "Metadata", "request_id": "req-1"}, {"type": "Flushed", "sequence_id": 0}], + logging_obj=logging_obj, + url_route="/deepgram/v1/speak", + result="websocket_connection_successful", + start_time=datetime.now(), + end_time=datetime.now(), + cache_hit=False, + request_body={WEBSOCKET_CLIENT_TEXT_COUNT_KEY: characters}, + passthrough_logging_payload=PassthroughStandardLoggingPayload( + url=AURA_2_URL, request_body={}, request_method="WEBSOCKET", cost_per_request=None + ), + litellm_params=litellm_params, + ) + + assert len(capturing_logger.payloads) == 1 + payload: Final = capturing_logger.payloads[0] + assert payload["model"] == "aura-2-thalia-en" + assert payload["custom_llm_provider"] == "deepgram" + assert payload["response_cost"] == pytest.approx(expected_cost) + assert payload["response_cost"] is not None + assert payload["metadata"]["user_api_key_team_id"] == "team-tts" + assert payload["id"] == "call-dg-speak-e2e" diff --git a/tests/unit/proxy/pass_through_endpoints/test_deepgram_ws_passthrough_routes.py b/tests/unit/proxy/pass_through_endpoints/test_deepgram_ws_passthrough_routes.py index 19418c4c349..b29e181b0be 100644 --- a/tests/unit/proxy/pass_through_endpoints/test_deepgram_ws_passthrough_routes.py +++ b/tests/unit/proxy/pass_through_endpoints/test_deepgram_ws_passthrough_routes.py @@ -1,7 +1,8 @@ -"""Deepgram ``/v1/listen`` passthrough WebSocket route: registration, auth, credential injection, target URL.""" +"""Deepgram ``/v1/listen`` and ``/v1/speak`` passthrough WebSocket routes: registration, auth, credential injection, +target URL.""" import asyncio -from collections.abc import Mapping +from collections.abc import Callable, Mapping from dataclasses import dataclass from types import MappingProxyType, SimpleNamespace from typing import Final @@ -15,12 +16,14 @@ from starlette.websockets import WebSocketDisconnect import litellm from litellm.caching.dual_cache import DualCache +from litellm.llms.deepgram.common_utils import deepgram_speak_characters from litellm.proxy._lazy_features import LAZY_FEATURES from litellm.proxy._types import LiteLLMRoutes, UserAPIKeyAuth from litellm.proxy.auth.auth_checks import cache_key_object from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import ( _websocket_relay, deepgram_listen_websocket_route, + deepgram_speak_websocket_route, router, ) from litellm.proxy.utils import hash_token @@ -30,6 +33,7 @@ GET_CREDENTIALS: Final = ( ) USER_API_KEY_AUTH: Final = "litellm.proxy.auth.user_api_key_auth.user_api_key_auth" LISTEN_PATHS: Final = ("/deepgram/v1/listen", "/deepgram/listen") +SPEAK_PATHS: Final = ("/deepgram/v1/speak", "/deepgram/speak") NOVA_2_STREAMING_KEY: Final = "deepgram/streaming/nova-2" pytestmark: Final = pytest.mark.usefixtures("local_model_cost_map") @@ -62,6 +66,7 @@ class _RelayCall: forward_headers: bool endpoint: str accept_websocket: bool + count_client_text: Callable[[str | bytes], int] | None = None class _FakeRelay: @@ -78,6 +83,7 @@ class _FakeRelay: forward_headers: bool, endpoint: str, accept_websocket: bool, + count_client_text: Callable[[str | bytes], int] | None = None, ) -> None: self.calls.append( _RelayCall( @@ -87,6 +93,7 @@ class _FakeRelay: forward_headers=forward_headers, endpoint=endpoint, accept_websocket=accept_websocket, + count_client_text=count_client_text, ) ) @@ -101,13 +108,13 @@ async def _serve(websocket: _FakeWebSocket, user_api_key_dict: UserAPIKeyAuth | return relay -def test_deepgram_listen_websocket_routes_registered(): +def test_deepgram_websocket_routes_registered() -> None: ws_paths = {route.path for route in router.routes if isinstance(route, WebSocketRoute)} - assert set(LISTEN_PATHS) <= ws_paths + assert {*LISTEN_PATHS, *SPEAK_PATHS} <= ws_paths -@pytest.mark.parametrize("path", LISTEN_PATHS) -def test_deepgram_listen_is_a_lazily_loaded_mapped_pass_through_route(path): +@pytest.mark.parametrize("path", (*LISTEN_PATHS, *SPEAK_PATHS)) +def test_deepgram_websocket_routes_are_lazily_loaded_mapped_pass_through_routes(path: str) -> None: """The route must be reachable before the passthrough module is imported and must be authed and billed as a mapped pass-through route like the other provider prefixes.""" feature = next(feature for feature in LAZY_FEATURES if feature.name == "llm_passthrough") @@ -474,14 +481,25 @@ def test_deepgram_listen_echoes_the_browser_subprotocol_that_carries_the_litellm assert [call.forward_headers for call in relay.calls] == [False] -def test_deepgram_listen_accepts_the_litellm_key_in_a_deepgram_sdk_token_header( - monkeypatch: pytest.MonkeyPatch, +@pytest.mark.parametrize( + ("url", "expected_target"), + [ + pytest.param("/deepgram/listen?model=nova-3", "wss://api.deepgram.com/v1/listen?model=nova-3", id="listen"), + pytest.param( + "/deepgram/speak?model=aura-2-thalia-en", + "wss://api.deepgram.com/v1/speak?model=aura-2-thalia-en", + id="speak", + ), + ], +) +def test_deepgram_routes_accept_the_litellm_key_in_a_deepgram_sdk_token_header( + url: str, expected_target: str, monkeypatch: pytest.MonkeyPatch ) -> None: """Deepgram SDKs send ``Authorization: Token ``, so pointing one at the proxy with a LiteLLM key as its API key must authenticate through the real key auth path and still send only the server's key upstream""" monkeypatch.delenv("DEEPGRAM_API_BASE", raising=False) monkeypatch.setattr(litellm, "max_budget", 0.0) - cache: Final = asyncio.run(_cache_restricted_key("sk-deepgram-sdk", ["nova-3"])) + cache: Final = asyncio.run(_cache_restricted_key("sk-deepgram-sdk", ["nova-3", "aura-2-thalia-en"])) relay: Final = _FakeRelay() client: Final = TestClient(_app_with_relay(relay)) @@ -496,13 +514,11 @@ def test_deepgram_listen_accepts_the_litellm_key_in_a_deepgram_sdk_token_header( llm_router=None, ), ): - with client.websocket_connect( - "/deepgram/listen?model=nova-3", headers={"Authorization": "Token sk-deepgram-sdk"} - ): + with client.websocket_connect(url, headers={"Authorization": "Token sk-deepgram-sdk"}): pass assert [(call.target, call.custom_headers) for call in relay.calls] == [ - ("wss://api.deepgram.com/v1/listen?model=nova-3", MappingProxyType({"Authorization": "Token dg-provider-key"})) + (expected_target, MappingProxyType({"Authorization": "Token dg-provider-key"})) ] @@ -518,3 +534,208 @@ def test_a_non_deepgram_websocket_route_still_refuses_a_token_header() -> None: assert disconnect.value.code == 1008 assert relay.calls == [] + + +async def _serve_speak(websocket: _FakeWebSocket, user_api_key_dict: UserAPIKeyAuth | None = None) -> _FakeRelay: + relay: Final = _FakeRelay() + await deepgram_speak_websocket_route( + websocket=websocket, + user_api_key_dict=user_api_key_dict or UserAPIKeyAuth(), + relay=relay, + ) + return relay + + +@pytest.mark.parametrize("path", SPEAK_PATHS) +async def test_deepgram_speak_forwards_query_injects_only_provider_auth_and_counts_characters( + path: str, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.delenv("DEEPGRAM_API_BASE", raising=False) + websocket: Final = _FakeWebSocket(path, "model=aura-2-thalia-en&encoding=linear16&sample_rate=24000") + caller: Final = UserAPIKeyAuth(api_key="sk-litellm-virtual", team_id="team-tts") + + with patch(GET_CREDENTIALS, return_value="dg-provider-key") as get_credentials: + relay: Final = await _serve_speak(websocket, caller) + + assert get_credentials.call_args.kwargs == {"custom_llm_provider": "deepgram", "region_name": None} + assert relay.calls == [ + _RelayCall( + target="wss://api.deepgram.com/v1/speak?model=aura-2-thalia-en&encoding=linear16&sample_rate=24000", + custom_headers=MappingProxyType({"Authorization": "Token dg-provider-key"}), + user_api_key_dict=caller, + forward_headers=False, + endpoint=path, + accept_websocket=False, + count_client_text=deepgram_speak_characters, + ) + ] + assert websocket.accepts == [None] + assert websocket.closed is None + + +async def test_deepgram_listen_does_not_count_client_text(monkeypatch: pytest.MonkeyPatch) -> None: + """Listen bills audio seconds from upstream frames, so it must keep handing the relay no counter""" + monkeypatch.delenv("DEEPGRAM_API_BASE", raising=False) + + with patch(GET_CREDENTIALS, return_value="dg-provider-key"): + relay: Final = await _serve(_FakeWebSocket("/deepgram/v1/listen", "model=nova-3")) + + assert [call.count_client_text for call in relay.calls] == [None] + + +@pytest.mark.parametrize( + ("query", "expected_target"), + [ + pytest.param("", "wss://api.deepgram.com/v1/speak?model=aura-asteria-en", id="no query"), + pytest.param("model=", "wss://api.deepgram.com/v1/speak?model=aura-asteria-en", id="blank model"), + pytest.param( + "encoding=mulaw&sample_rate=8000", + "wss://api.deepgram.com/v1/speak?encoding=mulaw&sample_rate=8000&model=aura-asteria-en", + id="other params kept", + ), + pytest.param( + "model=aura-2-thalia-en&model=aura-asteria-en", + "wss://api.deepgram.com/v1/speak?model=aura-2-thalia-en", + id="only the first model", + ), + ], +) +async def test_deepgram_speak_names_exactly_one_model_upstream( + query: str, expected_target: str, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.delenv("DEEPGRAM_API_BASE", raising=False) + + with patch(GET_CREDENTIALS, return_value="dg-provider-key"): + relay: Final = await _serve_speak(_FakeWebSocket("/deepgram/speak", query)) + + assert [call.target for call in relay.calls] == [expected_target] + + +async def test_deepgram_speak_honours_server_configured_api_base_and_ignores_the_callers( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The server's Deepgram key must only ever go to the server-configured host""" + monkeypatch.setenv("DEEPGRAM_API_BASE", "https://api.eu.deepgram.com/v1/") + websocket: Final = _FakeWebSocket( + "/deepgram/v1/speak", "api_base=wss%3A%2F%2Fattacker.example%2Fv1&model=aura-2-thalia-en" + ) + + with patch(GET_CREDENTIALS, return_value="dg-provider-key"): + relay: Final = await _serve_speak(websocket) + + assert [call.target for call in relay.calls] == [ + "wss://api.eu.deepgram.com/v1/speak?api_base=wss%3A%2F%2Fattacker.example%2Fv1&model=aura-2-thalia-en" + ] + + +async def test_deepgram_speak_closes_before_accepting_when_provider_credentials_missing() -> None: + websocket: Final = _FakeWebSocket("/deepgram/v1/speak", "model=aura-2-thalia-en") + + with patch(GET_CREDENTIALS, return_value=None): + relay: Final = await _serve_speak(websocket) + + assert websocket.closed is not None + assert websocket.closed[0] == 1011 + assert "DEEPGRAM_API_KEY" in websocket.closed[1] + assert websocket.accepts == [] + assert relay.calls == [] + + +@pytest.mark.parametrize( + ("query", "missing_key", "removed_row"), + [ + pytest.param("model=flux-tts-en", "deepgram/flux-tts-en", None, id="voice outside the priced families"), + pytest.param("model=aura-3-x-en", "deepgram/aura-3-x-en", None, id="unknown aura generation"), + pytest.param("model=aura-x", "deepgram/aura-x", None, id="aura voice without a language"), + pytest.param("model=nova-3", "deepgram/nova-3", None, id="transcription model with a per-second row"), + pytest.param("model=aura-2-thalia-en", "deepgram/aura-2", "deepgram/aura-2", id="family row removed"), + ], +) +async def test_deepgram_speak_refuses_sessions_it_cannot_price( + query: str, missing_key: str, removed_row: str | None, monkeypatch: pytest.MonkeyPatch +) -> None: + """A session without a per-character price would be logged at no cost, so the proxy closes it before Deepgram is + contacted and names the registry row to add""" + monkeypatch.delenv("DEEPGRAM_API_BASE", raising=False) + if removed_row is not None: + monkeypatch.delitem(litellm.model_cost, removed_row) + websocket: Final = _FakeWebSocket("/deepgram/v1/speak", query) + + with patch(GET_CREDENTIALS, return_value="dg-provider-key"): + relay: Final = await _serve_speak(websocket) + + assert relay.calls == [] + assert websocket.accepts == [None] + assert websocket.closed is not None + assert websocket.closed[0] == 1008 + assert missing_key in websocket.closed[1] + assert "dg-provider-key" not in websocket.closed[1] + + +def test_deepgram_speak_rejects_connections_without_a_litellm_key() -> None: + relay: Final = _FakeRelay() + client: Final = TestClient(_app_with_relay(relay)) + + with patch(GET_CREDENTIALS, return_value="dg-provider-key") as get_credentials: + with pytest.raises(WebSocketDisconnect) as disconnect: + with client.websocket_connect("/deepgram/v1/speak?model=aura-2-thalia-en"): + pass + + assert disconnect.value.code == 1008 + assert relay.calls == [] + get_credentials.assert_not_called() + + +@pytest.mark.parametrize( + ("query", "expected_target"), + [ + pytest.param( + "model=aura-2-thalia-en", + "wss://api.deepgram.com/v1/speak?model=aura-2-thalia-en", + id="allowed voice named", + ), + pytest.param( + "model=aura-2-thalia-en&model=aura-2-zeus-en", + "wss://api.deepgram.com/v1/speak?model=aura-2-thalia-en", + id="second voice stripped", + ), + pytest.param("model=aura-2-zeus-en", None, id="denied voice named"), + pytest.param("", None, id="voice omitted, default denied"), + pytest.param("model=&encoding=mulaw", None, id="voice blank, default denied"), + ], +) +def test_deepgram_speak_authorizes_the_voice_it_will_actually_send_upstream( + query: str, expected_target: str | None, monkeypatch: pytest.MonkeyPatch +) -> None: + """A key allowed only one voice must not reach another by leaving ``model`` out or naming a second one, because + the real key auth path must see the same model the upstream target carries""" + monkeypatch.delenv("DEEPGRAM_API_BASE", raising=False) + monkeypatch.setattr(litellm, "max_budget", 0.0) + cache: Final = asyncio.run(_cache_restricted_key("sk-only-thalia", ["aura-2-thalia-en"])) + relay: Final = _FakeRelay() + client: Final = TestClient(_app_with_relay(relay)) + url: Final = f"/deepgram/v1/speak?{query}" + headers: Final = {"Authorization": "Bearer sk-only-thalia"} + + with ( + patch(GET_CREDENTIALS, return_value="dg-provider-key"), + patch.multiple( # test-quality-ok: the real key auth path reads these proxy_server globals and has no injection seam + "litellm.proxy.proxy_server", + master_key="sk-master", + prisma_client=MagicMock(), + user_api_key_cache=cache, + llm_model_list=None, + llm_router=None, + ), + ): + if expected_target is not None: + with client.websocket_connect(url, headers=headers): + pass + assert [call.target for call in relay.calls] == [expected_target] + return + with pytest.raises(WebSocketDisconnect) as disconnect: + with client.websocket_connect(url, headers=headers): + pass + + assert disconnect.value.code == 1008 + assert relay.calls == [] diff --git a/tests/unit/proxy/pass_through_endpoints/test_pass_through_endpoints.py b/tests/unit/proxy/pass_through_endpoints/test_pass_through_endpoints.py index 466c3be5114..5c879f9b97b 100644 --- a/tests/unit/proxy/pass_through_endpoints/test_pass_through_endpoints.py +++ b/tests/unit/proxy/pass_through_endpoints/test_pass_through_endpoints.py @@ -35,6 +35,7 @@ from litellm.constants import DEFAULT_REQUEST_TIMEOUT_SECONDS from litellm.integrations.custom_guardrail import CustomGuardrail from litellm.integrations.custom_logger import CustomLogger from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj +from litellm.llms.deepgram.common_utils import deepgram_speak_characters from litellm.proxy._lazy_features import LazyFeature, attach_lazy_features from litellm.proxy._types import ProxyException, UserAPIKeyAuth from litellm.proxy.litellm_pre_call_utils import LiteLLMProxyRequestSetup @@ -7651,10 +7652,11 @@ class _RelayedSession: async def _relay_and_capture_spend( monkeypatch: pytest.MonkeyPatch, - upstream_ws: FakeUpstreamWebSocket, + upstream_ws: "FakeUpstreamWebSocket | SpeakingUpstreamWebSocket", websocket: MagicMock, target: str, endpoint: str, + count_client_text: Callable[[str | bytes], int] | None = None, ) -> _RelayedSession: """Runs the relay with the real success handler and returns what a spend logger receives""" capturing_logger: Final = _CapturingSpendLogger() @@ -7685,6 +7687,7 @@ async def _relay_and_capture_spend( forward_headers=False, endpoint=endpoint, accept_websocket=False, + count_client_text=count_client_text, ), timeout=RELAY_HANG_GUARD_SECONDS, ) @@ -7822,6 +7825,174 @@ async def _track_cost_for_passthrough_kwargs(kwargs: dict) -> AsyncMock: return increment_spend_counters +DEEPGRAM_SPEAK_TARGET: Final = "wss://api.deepgram.com/v1/speak?model=aura-2-thalia-en" +SPEAK_AUDIO_CHUNK: Final = bytes(range(256)) * 8 +SPOKEN_TEXT: Final = "The quick brown fox jumps over the lazy dog." +SPEAK_FRAME: Final = json.dumps({"type": "Speak", "text": SPOKEN_TEXT}) +FLUSH_FRAME: Final = json.dumps({"type": "Flush"}) +FLUSHED_FRAME: Final = json.dumps({"type": "Flushed", "sequence_id": 0}) +SPEAK_CLOSE_FRAME: Final = json.dumps({"type": "Close"}) +CLIENT_BINARY_FRAME: Final = b"\x00\xffclient-binary" +SPEAK_SESSION_FRAMES: Final = ({"text": SPEAK_FRAME}, {"bytes": CLIENT_BINARY_FRAME}, {"text": FLUSH_FRAME}) +SPEAK_SESSION_CLIENT_LENGTH: Final = sum(map(len, (SPEAK_FRAME, CLIENT_BINARY_FRAME, FLUSH_FRAME, SPEAK_CLOSE_FRAME))) + + +class SpeakingUpstreamWebSocket: + """Answers each Speak with audio as soon as it arrives and ends the session on Close, the way Deepgram speak + does, so a relay that held audio back until the end would never see the Close it waits for""" + + def __init__(self, close_code: int = 1000, close_reason: str = "") -> None: + self.received: list[str | bytes] = [] + self.close = AsyncMock() + self._close_code = close_code + self._close_reason = close_reason + self._replies: asyncio.Queue[str | bytes | None] = asyncio.Queue() + + async def send(self, message: str | bytes) -> None: + self.received.append(message) + match json.loads(message).get("type") if isinstance(message, str) else None: + case "Speak": + self._replies.put_nowait(SPEAK_AUDIO_CHUNK) + case "Flush": + self._replies.put_nowait(FLUSHED_FRAME) + case "Close": + self._replies.put_nowait(None) + + async def recv(self, decode: bool | None = None) -> str | bytes: + from websockets.exceptions import ConnectionClosedError, ConnectionClosedOK + from websockets.frames import Close + + reply: Final = await self._replies.get() + if reply is not None: + return reply + close: Final = Close(self._close_code, self._close_reason) + closed: Final = ConnectionClosedOK if self._close_code == 1000 else ConnectionClosedError + raise closed(rcvd=close, sent=close, rcvd_then_sent=True) + + +def _speak_client(frames: tuple[dict[str, str | bytes], ...] = SPEAK_SESSION_FRAMES) -> MagicMock: + """A client socket that sends ``frames``, then Close once the first audio has reached it""" + first_audio: Final = asyncio.Event() + + async def script() -> AsyncIterator[dict[str, str | bytes]]: + for frame in frames: + yield {"type": "websocket.receive", **frame} + await first_audio.wait() + yield {"type": "websocket.receive", "text": SPEAK_CLOSE_FRAME} + await asyncio.Event().wait() + + steps: Final = script() + + async def receive() -> dict[str, str | bytes]: + return await anext(steps) + + websocket: Final = _client_websocket(receive) + websocket.send_bytes = AsyncMock(side_effect=lambda data: first_audio.set()) + return websocket + + +def _aura_2_usd_per_character() -> float: + return litellm.model_cost["deepgram/aura-2"]["input_cost_per_character"] + + +@pytest.mark.asyncio +@pytest.mark.usefixtures("local_model_cost_map") +async def test_websocket_passthrough_counts_client_text_without_changing_what_is_relayed( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Audio reaches the client while the session is open, every client frame reaches the upstream unchanged, and + the session is billed for the summed count of the text and binary frames""" + upstream_ws: Final = SpeakingUpstreamWebSocket() + websocket: Final = _speak_client() + + session: Final = await _relay_and_capture_spend( + monkeypatch, upstream_ws, websocket, DEEPGRAM_SPEAK_TARGET, "/deepgram/v1/speak", count_client_text=len + ) + + assert upstream_ws.received == [SPEAK_FRAME, CLIENT_BINARY_FRAME, FLUSH_FRAME, SPEAK_CLOSE_FRAME] + assert [call.args[0] for call in websocket.send_bytes.await_args_list] == [SPEAK_AUDIO_CHUNK] + assert [call.args[0] for call in websocket.send_text.await_args_list] == [FLUSHED_FRAME] + assert [log["response_cost"] for log in session.spend_logs] == [ + pytest.approx(SPEAK_SESSION_CLIENT_LENGTH * _aura_2_usd_per_character()) + ] + websocket.close.assert_awaited_once_with() + + +@pytest.mark.asyncio +@pytest.mark.usefixtures("local_model_cost_map") +@pytest.mark.parametrize( + ("close_code", "close_reason"), + [ + pytest.param(1008, "DATA-0000: invalid text", id="policy"), + pytest.param(1011, "NET-0001: internal", id="internal"), + ], +) +async def test_websocket_passthrough_counts_client_text_when_the_upstream_ends_with_an_error_close( + monkeypatch: pytest.MonkeyPatch, close_code: int, close_reason: str +) -> None: + upstream_ws: Final = SpeakingUpstreamWebSocket(close_code=close_code, close_reason=close_reason) + websocket: Final = _speak_client() + + session: Final = await _relay_and_capture_spend( + monkeypatch, upstream_ws, websocket, DEEPGRAM_SPEAK_TARGET, "/deepgram/v1/speak", count_client_text=len + ) + + websocket.close.assert_awaited_once_with(code=close_code, reason=close_reason) + assert [log["response_cost"] for log in session.spend_logs] == [ + pytest.approx(SPEAK_SESSION_CLIENT_LENGTH * _aura_2_usd_per_character()) + ] + + +class _DrainStallingUpstreamWebSocket(SpeakingUpstreamWebSocket): + """Takes the client's Close, then stalls in send the way a full write buffer does, so the session ends while the + relay is still inside that send""" + + async def send(self, message: str | bytes) -> None: + await super().send(message) + if message == SPEAK_CLOSE_FRAME: + await asyncio.Event().wait() + + +@pytest.mark.asyncio +@pytest.mark.usefixtures("local_model_cost_map") +async def test_websocket_passthrough_bills_a_client_frame_whose_send_is_cancelled_after_reaching_the_upstream( + monkeypatch: pytest.MonkeyPatch, +) -> None: + session: Final = await _relay_and_capture_spend( + monkeypatch, + _DrainStallingUpstreamWebSocket(), + _speak_client(), + DEEPGRAM_SPEAK_TARGET, + "/deepgram/v1/speak", + count_client_text=len, + ) + + assert [log["response_cost"] for log in session.spend_logs] == [ + pytest.approx(SPEAK_SESSION_CLIENT_LENGTH * _aura_2_usd_per_character()) + ] + + +@pytest.mark.asyncio +@pytest.mark.usefixtures("local_model_cost_map") +async def test_deepgram_speak_session_logs_character_spend_once_and_never_the_spoken_text( + monkeypatch: pytest.MonkeyPatch, +) -> None: + session: Final = await _relay_and_capture_spend( + monkeypatch, + SpeakingUpstreamWebSocket(), + _speak_client(), + DEEPGRAM_SPEAK_TARGET, + "/deepgram/v1/speak", + count_client_text=deepgram_speak_characters, + ) + + assert [(log["model"], log["custom_llm_provider"]) for log in session.spend_logs] == [ + ("aura-2-thalia-en", "deepgram") + ] + assert session.spend_logs[0]["response_cost"] == pytest.approx(len(SPOKEN_TEXT) * _aura_2_usd_per_character()) + assert SPOKEN_TEXT not in json.dumps(session.spend_logs[0], default=str) + + @pytest.mark.asyncio async def test_passthrough_success_reconciles_budget_reservation(): """ diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py index 6c7f6a9d869..484f62bba2c 100644 --- a/tests/unit/test_utils.py +++ b/tests/unit/test_utils.py @@ -1250,6 +1250,7 @@ _COST_MAP_ROW_SCHEMA: Final = { "/v1/videos", "/vertex_ai/live", "/v1/listen", + "/v1/speak", "/v1/systemone", "/v1/decisions", "/v1beta/interactions",