diff --git a/litellm/constants.py b/litellm/constants.py index 1d5c6404bfd..a6c73dc0eec 100644 --- a/litellm/constants.py +++ b/litellm/constants.py @@ -423,6 +423,8 @@ 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" +# https://developers.deepgram.com/reference/text-to-speech/speak-streaming, checked 2026-10-28 +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..0b8f1d7f969 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 deepgram_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 deepgram_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 deepgram_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 deepgram_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, ...]: @@ -199,3 +221,34 @@ 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_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_upstream_model(upstream_url, DEEPGRAM_SPEAK_DEFAULT_MODEL) + 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") + + +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 8719565a96f..6ae3a405039 100644 --- a/litellm/model_prices_and_context_window_backup.json +++ b/litellm/model_prices_and_context_window_backup.json @@ -22775,6 +22775,30 @@ "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" + }, + "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" + }, + "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 e478096f7ad..2e89996c4e3 100644 --- a/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py @@ -44,6 +44,7 @@ from litellm.constants import ( AZURE_SPEECH_STT_DOMAIN, AZURE_SPEECH_SUBSCRIPTION_KEY_HEADER, BEDROCK_AGENT_RUNTIME_PASS_THROUGH_ROUTES, + DEEPGRAM_SPEAK_DEFAULT_MODEL, PASS_THROUGH_HEADER_PREFIX, TYPESAFE_DANGER_ZONE_HEADER_PREFIX, ) @@ -63,6 +64,11 @@ from litellm.llms.deepgram.common_utils import ( deepgram_listen_registry_key, deepgram_listen_requested_model, deepgram_listen_websocket_target, + deepgram_requested_model, + deepgram_speak_character_rate, + deepgram_speak_characters, + deepgram_speak_registry_key, + deepgram_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 @@ -3354,6 +3360,7 @@ class _WebsocketRelay(Protocol): forward_headers: bool, endpoint: str, accept_websocket: bool, + count_client_text: Callable[[str | bytes], int] | None = None, ) -> None: ... @@ -3469,12 +3476,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, @@ -3485,23 +3502,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( @@ -3512,6 +3514,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_requested_model(websocket.url.query, DEEPGRAM_SPEAK_DEFAULT_MODEL) + ) + + +def _deepgram_speak_refusal(target: str) -> str | None: + if deepgram_speak_character_rate(target) is not None: + 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_websocket_target( + get_secret_str("DEEPGRAM_API_BASE"), "speak", websocket.url.query, DEEPGRAM_SPEAK_DEFAULT_MODEL + ) + 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..0473c527a2a --- /dev/null +++ b/litellm/proxy/pass_through_endpoints/llm_provider_handlers/deepgram_speak_passthrough_logging_handler.py @@ -0,0 +1,35 @@ +import json +from typing import Final +from urllib.parse import urlparse + +import litellm +from litellm.constants import DEEPGRAM_SPEAK_DEFAULT_MODEL +from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj +from litellm.llms.deepgram.common_utils import deepgram_speak_character_rate, deepgram_upstream_model +from litellm.types.passthrough_endpoints.pass_through_endpoints import WEBSOCKET_CLIENT_TEXT_COUNT_KEY +from litellm.types.utils import StandardPassThroughResponseObject + + +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("/speak") + + @staticmethod + def deepgram_speak_passthrough_handler( + logging_obj: LiteLLMLoggingObj, upstream_url: str + ) -> StandardPassThroughResponseObject: + counted: Final = logging_obj.model_call_details.get(WEBSOCKET_CLIENT_TEXT_COUNT_KEY) + characters: Final = counted if isinstance(counted, int) else 0 + rate: Final = deepgram_speak_character_rate(upstream_url) + model: Final = deepgram_upstream_model(upstream_url, DEEPGRAM_SPEAK_DEFAULT_MODEL) + 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"] = litellm.LlmProviders.DEEPGRAM.value # rebind-ok: same + logging_obj.model_call_details["response_cost"] = None if rate is None else characters * rate # rebind-ok: same + return StandardPassThroughResponseObject(response=json.dumps({"characters": characters})) diff --git a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py index e017eee3d27..b24d3f107e4 100644 --- a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py @@ -140,6 +140,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, ) @@ -2508,6 +2509,7 @@ async def websocket_passthrough_request( 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. @@ -2521,6 +2523,10 @@ 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 logging_obj.model_call_details, 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 @@ -2531,6 +2537,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) @@ -2647,6 +2654,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: @@ -2717,9 +2734,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: @@ -2845,6 +2862,8 @@ async def websocket_passthrough_request( mock_response: Final = MockWebSocketResponse(target) + if count_client_text is not None: + logging_obj.model_call_details[WEBSOCKET_CLIENT_TEXT_COUNT_KEY] = client_text_count # Use the same success handler as HTTP passthrough endpoints GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue( async_coroutine=pass_through_endpoint_logging.pass_through_async_success_handler( diff --git a/litellm/proxy/pass_through_endpoints/success_handler.py b/litellm/proxy/pass_through_endpoints/success_handler.py index 9c6c3111209..c7a7d0d0808 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,12 @@ 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): + standard_logging_response_object = ( # rebind-ok: elif-chain + DeepgramSpeakPassthroughLoggingHandler.deepgram_speak_passthrough_handler( + logging_obj=logging_obj, upstream_url=str(httpx_response.request.url) + ) + ) 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 8719565a96f..6ae3a405039 100644 --- a/model_prices_and_context_window.json +++ b/model_prices_and_context_window.json @@ -22775,6 +22775,30 @@ "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" + }, + "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" + }, + "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..45f4722ecd8 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 @@ -6,6 +7,7 @@ from urllib.parse import parse_qs, urlparse import pytest import litellm +from litellm.constants import DEEPGRAM_SPEAK_DEFAULT_MODEL from litellm.llms.deepgram.common_utils import ( deepgram_listen_addon_pricing_models, deepgram_listen_audio_seconds, @@ -18,6 +20,12 @@ from litellm.llms.deepgram.common_utils import ( deepgram_listen_requested_model, deepgram_listen_transcript, deepgram_listen_websocket_target, + deepgram_requested_model, + deepgram_speak_character_rate, + deepgram_speak_characters, + deepgram_speak_registry_key, + deepgram_upstream_model, + deepgram_websocket_target, ) NOVA_3_URL: Final = "wss://api.deepgram.com/v1/listen?model=nova-3&encoding=linear16&sample_rate=16000" @@ -324,3 +332,133 @@ 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_websocket_target(api_base, "speak", query_string, DEEPGRAM_SPEAK_DEFAULT_MODEL) == 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_requested_model, so the upstream must use exactly that voice""" + requested: Final = deepgram_requested_model(query_string, DEEPGRAM_SPEAK_DEFAULT_MODEL) + target: Final = deepgram_websocket_target(None, "speak", query_string, DEEPGRAM_SPEAK_DEFAULT_MODEL) + + assert parse_qs(urlparse(target).query)["model"] == [requested] + assert deepgram_upstream_model(target, DEEPGRAM_SPEAK_DEFAULT_MODEL) == requested + + +@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-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: + assert deepgram_speak_registry_key(f"{SPEAK_BASE}?model={model}") == f"deepgram/{expected_family}" + + +@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 + + +@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..b2c821cf9bf --- /dev/null +++ b/tests/unit/proxy/pass_through_endpoints/llm_provider_handlers/test_deepgram_speak_passthrough_logging_handler.py @@ -0,0 +1,89 @@ +"""Deepgram ``/v1/speak`` WebSocket passthrough: per-character cost tracking from the relay's client text count.""" + +import json +from datetime import datetime +from typing import Final + +import pytest + +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.types.passthrough_endpoints.pass_through_endpoints import WEBSOCKET_CLIENT_TEXT_COUNT_KEY + +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(characters: object) -> LiteLLMLoggingObj: + logging_obj: Final = LiteLLMLoggingObj( + model="unknown", + messages=[{"role": "user", "content": "WebSocket connection"}], + stream=True, + call_type="pass_through_endpoint", + start_time=datetime.now(), + litellm_call_id="call-dg-speak", + function_id="websocket_passthrough", + ) + logging_obj.model_call_details[WEBSOCKET_CLIENT_TEXT_COUNT_KEY] = characters + return logging_obj + + +def _billed_cost(upstream_url: str, characters: object) -> object: + logging_obj: Final = _logging_obj(characters) + DeepgramSpeakPassthroughLoggingHandler.deepgram_speak_passthrough_handler(logging_obj, upstream_url) + return logging_obj.model_call_details["response_cost"] + + +def test_handler_bills_spoken_characters_at_the_aura_2_rate_and_logs_only_the_count() -> None: + logging_obj: Final = _logging_obj(SPOKEN_CHARACTERS) + + result: Final = DeepgramSpeakPassthroughLoggingHandler.deepgram_speak_passthrough_handler(logging_obj, AURA_2_URL) + + assert logging_obj.model_call_details["response_cost"] == pytest.approx( + SPOKEN_CHARACTERS * AURA_2_USD_PER_CHARACTER + ) + assert logging_obj.model == logging_obj.model_call_details["model"] == "aura-2-thalia-en" + assert logging_obj.model_call_details["custom_llm_provider"] == "deepgram" + assert result == {"response": json.dumps({"characters": SPOKEN_CHARACTERS})} + + +def test_handler_bills_an_aura_1_voice_at_the_aura_1_rate() -> None: + assert _billed_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 _billed_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 _billed_cost(f"wss://api.deepgram.com/v1/speak?model={model}", 1000) is None 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 404ed0f5800..5651cf154f2 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") @@ -472,3 +479,180 @@ def test_deepgram_listen_echoes_the_browser_subprotocol_that_carries_the_litellm MappingProxyType({"Authorization": "Token dg-provider-key"}) ] assert [call.forward_headers for call in relay.calls] == [False] + + +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] + + +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 b0504af1b5e..8ce243d1536 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 @@ -7,7 +7,7 @@ import os import sys import uuid import zlib -from collections.abc import AsyncIterator, Callable, Mapping +from collections.abc import AsyncIterator, Callable, Coroutine, Mapping from contextlib import ExitStack, contextmanager from dataclasses import dataclass from datetime import datetime @@ -34,6 +34,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 @@ -69,6 +70,7 @@ from litellm.types.passthrough_endpoints.pass_through_endpoints import ( EndpointType, PassthroughStandardLoggingPayload, ) +from litellm.types.utils import StandardLoggingPayload from tests._master_key import MASTER_KEY from tests._vcr_conftest_common import install_live_call_probe, record_vcr_outcome @@ -7715,6 +7717,66 @@ async def test_websocket_passthrough_sends_deepgram_audio_bytes_and_control_text upstream_ws.close.assert_awaited_once() +class _CapturingSpendLogger(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"]) + + +@dataclass(frozen=True, slots=True) +class _RelayedSession: + spend_logs: tuple[StandardLoggingPayload, ...] + failure_hook: AsyncMock + + +async def _relay_and_capture_spend( + monkeypatch: pytest.MonkeyPatch, + 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() + monkeypatch.setattr(litellm, "_async_success_callback", [capturing_logger]) + monkeypatch.setattr(litellm, "success_callback", []) + monkeypatch.setattr(litellm, "callbacks", []) + queued: Final[list[Coroutine[object, object, None]]] = [] # mutable-ok: the fake worker keeps each coroutine + with ( + patch("litellm.proxy.proxy_server.proxy_logging_obj") as proxy_logging, + patch( + "litellm.proxy.pass_through_endpoints.pass_through_endpoints.connect", + return_value=FakeUpstreamConnect(upstream_ws), + ), + patch("litellm.proxy.pass_through_endpoints.pass_through_endpoints.GLOBAL_LOGGING_WORKER") as worker, + ): + proxy_logging.pre_call_hook = AsyncMock(return_value={}) + proxy_logging.post_call_success_hook = AsyncMock() + proxy_logging.post_call_failure_hook = AsyncMock() + worker.ensure_initialized_and_enqueue = MagicMock( + side_effect=lambda async_coroutine: queued.append(async_coroutine) + ) + await websocket_passthrough_request( + websocket=websocket, + target=target, + custom_headers={"Authorization": "Token dg-provider-key"}, + user_api_key_dict=UserAPIKeyAuth(), + forward_headers=False, + endpoint=endpoint, + accept_websocket=False, + count_client_text=count_client_text, + ) + for coroutine in queued: + await coroutine + return _RelayedSession( + spend_logs=tuple(capturing_logger.payloads), failure_hook=proxy_logging.post_call_failure_hook + ) + + @pytest.mark.asyncio async def test_websocket_passthrough_vertex_live_setup_ack_names_the_model_but_is_not_billed_as_usage(): """Vertex Live keeps its special first frame: the setup acknowledgement is forwarded verbatim, read for the @@ -7808,6 +7870,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 146e2df6413..07747ac5702 100644 --- a/tests/unit/test_utils.py +++ b/tests/unit/test_utils.py @@ -1252,6 +1252,7 @@ _COST_MAP_ROW_SCHEMA: Final = { "/v1/videos", "/vertex_ai/live", "/v1/listen", + "/v1/speak", "/v1/systemone", "/v1/decisions", "/v1beta/interactions",