From 763cb04df2a3c0153bae62c7cf61922dfbaa7df2 Mon Sep 17 00:00:00 2001 From: shrey kharbanda Date: Mon, 26 Oct 2026 12:20:32 -0700 Subject: [PATCH 01/10] chore(cost-map): add Deepgram Aura-2 and Aura-1 text-to-speech rates Deepgram bills streaming text-to-speech per input character, at $0.030 per 1k characters for Aura-2 voices and $0.0150 per 1k for Aura-1 voices on pay as you go (https://deepgram.com/pricing). One row per voice family keeps the map small, since every voice in a family costs the same The upcoming /deepgram/v1/speak WebSocket passthrough prices sessions from these rows and refuses voices that have none Refs LIT-10070 --- ...odel_prices_and_context_window_backup.json | 26 +++++++++++++++++++ model_prices_and_context_window.json | 26 +++++++++++++++++++ .../deepgram/test_deepgram_common_utils.py | 20 ++++++++++++++ tests/unit/test_utils.py | 1 + 4 files changed, 73 insertions(+) diff --git a/litellm/model_prices_and_context_window_backup.json b/litellm/model_prices_and_context_window_backup.json index 1f2bd7d7837..446af103664 100644 --- a/litellm/model_prices_and_context_window_backup.json +++ b/litellm/model_prices_and_context_window_backup.json @@ -22775,6 +22775,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/model_prices_and_context_window.json b/model_prices_and_context_window.json index 1f2bd7d7837..446af103664 100644 --- a/model_prices_and_context_window.json +++ b/model_prices_and_context_window.json @@ -22775,6 +22775,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..7f480d3459f 100644 --- a/tests/unit/llms/deepgram/test_deepgram_common_utils.py +++ b/tests/unit/llms/deepgram/test_deepgram_common_utils.py @@ -324,3 +324,23 @@ 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 +): + """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 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", From 4ae796053f21f02f5bd7250ded80cda851f5284b Mon Sep 17 00:00:00 2001 From: shrey kharbanda Date: Mon, 26 Oct 2026 13:41:38 -0700 Subject: [PATCH 02/10] feat(passthrough): let a WebSocket route count client text for billing websocket_passthrough_request takes an optional count_client_text callback. It is applied to each client frame, text or binary, after the frame reaches the upstream, and only the summed integer reaches the success handler, under WEBSOCKET_CLIENT_TEXT_COUNT_KEY in request_body. The frames are never kept, so a route can bill on client text it must not store. Binary frames are counted too, since an upstream may parse them the same way as text frames Frames are relayed unchanged in both directions. Callers that pass no counter keep request_body={}. custom_headers is typed as the read-only Mapping[str, str] the relay actually uses Refs LIT-10070 --- .../pass_through_endpoints.py | 17 +- .../pass_through_endpoints.py | 2 + .../test_pass_through_endpoints.py | 146 ++++++++++++++++++ 3 files changed, 163 insertions(+), 2 deletions(-) diff --git a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py index 650dcae3239..09652dfca2a 100644 --- a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py @@ -139,6 +139,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, ) @@ -2489,13 +2490,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. @@ -2509,6 +2511,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 @@ -2519,6 +2524,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) @@ -2637,6 +2643,7 @@ async def websocket_passthrough_request( async def forward_client_to_upstream() -> None: """Forward messages from client to upstream WebSocket""" + nonlocal client_text_count try: while True: message = await websocket.receive() @@ -2706,8 +2713,12 @@ 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)) + if count_client_text is not None: + client_text_count += count_client_text(text_data) elif bytes_data is not None: await upstream_ws.send(bytes_data) + if count_client_text is not None: + client_text_count += count_client_text(bytes_data) except asyncio.CancelledError: raise except Exception: @@ -2844,7 +2855,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/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/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..a797f8084b9 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 @@ -66,6 +66,7 @@ from litellm.types.guardrails import GuardrailEventHooks from litellm.types.passthrough_endpoints.pass_through_endpoints import ( LITELLM_PASS_THROUGH_DEPLOYMENT_MODEL_INFO_STATE_KEY, LITELLM_PASS_THROUGH_RAW_BODY_STATE_KEY, + WEBSOCKET_CLIENT_TEXT_COUNT_KEY, EndpointType, PassthroughStandardLoggingPayload, ) @@ -7808,6 +7809,151 @@ async def _track_cost_for_passthrough_kwargs(kwargs: dict) -> AsyncMock: return increment_spend_counters +DEEPGRAM_SPEAK_TARGET = "wss://api.deepgram.com/v1/speak?model=aura-2-thalia-en" +SPEAK_AUDIO_CHUNK = bytes(range(256)) * 8 +SPOKEN_TEXT = "The quick brown fox jumps over the lazy dog." +SPEAK_FRAME = json.dumps({"type": "Speak", "text": SPOKEN_TEXT}) +FLUSH_FRAME = json.dumps({"type": "Flush"}) +FLUSHED_FRAME = json.dumps({"type": "Flushed", "sequence_id": 0}) +SPEAK_CLOSE_FRAME = json.dumps({"type": "Close"}) +CLIENT_BINARY_FRAME = b"\x00\xffclient-binary" +SPEAK_SESSION_FRAMES = ({"text": SPEAK_FRAME}, {"bytes": CLIENT_BINARY_FRAME}, {"text": FLUSH_FRAME}) +SPEAK_SESSION_CLIENT_LENGTH = sum(map(len, (SPEAK_FRAME, CLIENT_BINARY_FRAME, FLUSH_FRAME, SPEAK_CLOSE_FRAME))) +RELAY_HANG_GUARD_SECONDS = 5.0 + + +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 = ""): + 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 = await self._replies.get() + if reply is not None: + return reply + close = Close(self._close_code, self._close_reason) + closed = 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): + """A client socket that sends ``frames``, then Close once the first audio has reached it""" + first_audio = asyncio.Event() + + async def script(): + 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 = script() + + async def receive(): + return await anext(steps) + + websocket = _client_websocket(receive) + websocket.send_bytes = AsyncMock(side_effect=lambda data: first_audio.set()) + return websocket + + +async def _relay_deepgram_speak(upstream_ws, websocket, count_client_text): + """Runs the generic relay the way the Deepgram speak route does and returns the success handler mock""" + with ( + _patched_websocket_passthrough_environment(upstream_ws), + patch( # test-quality-ok: pass_through_endpoint_logging is a module global read inside websocket_passthrough_request; there is no injection seam + "litellm.proxy.pass_through_endpoints.pass_through_endpoints." + "pass_through_endpoint_logging.pass_through_async_success_handler", + new=AsyncMock(), + ) as success_handler, + ): + await asyncio.wait_for( + websocket_passthrough_request( + websocket=websocket, + target=DEEPGRAM_SPEAK_TARGET, + custom_headers={"Authorization": "Token dg-provider-key"}, + user_api_key_dict=UserAPIKeyAuth(), + forward_headers=False, + endpoint="/deepgram/v1/speak", + accept_websocket=False, + count_client_text=count_client_text, + ), + timeout=RELAY_HANG_GUARD_SECONDS, + ) + return success_handler + + +@pytest.mark.asyncio +async def test_websocket_passthrough_counts_client_text_without_changing_what_is_relayed(): + """Audio reaches the client while the session is open, every client frame reaches the upstream unchanged, and + the success handler gets only the summed count of the text and binary frames, never the frames""" + upstream_ws = SpeakingUpstreamWebSocket() + websocket = _speak_client() + + success_handler = await _relay_deepgram_speak(upstream_ws, websocket, 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 success_handler.call_args.kwargs["request_body"] == { + WEBSOCKET_CLIENT_TEXT_COUNT_KEY: SPEAK_SESSION_CLIENT_LENGTH + } + websocket.close.assert_awaited_once_with() + + +@pytest.mark.asyncio +async def test_websocket_passthrough_without_a_counter_reports_an_empty_request_body(): + """Listen, realtime and every other caller pass no counter, so their success handler sees what it saw before""" + upstream_ws = SpeakingUpstreamWebSocket() + + success_handler = await _relay_deepgram_speak(upstream_ws, _speak_client(), count_client_text=None) + + assert upstream_ws.received == [SPEAK_FRAME, CLIENT_BINARY_FRAME, FLUSH_FRAME, SPEAK_CLOSE_FRAME] + assert success_handler.call_args.kwargs["request_body"] == {} + assert success_handler.call_args.kwargs["response_body"] == [json.loads(FLUSHED_FRAME)] + + +@pytest.mark.asyncio +@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( + close_code: int, close_reason: str +): + upstream_ws = SpeakingUpstreamWebSocket(close_code=close_code, close_reason=close_reason) + websocket = _speak_client() + + success_handler = await _relay_deepgram_speak(upstream_ws, websocket, count_client_text=len) + + websocket.close.assert_awaited_once_with(code=close_code, reason=close_reason) + assert success_handler.call_args.kwargs["request_body"] == { + WEBSOCKET_CLIENT_TEXT_COUNT_KEY: SPEAK_SESSION_CLIENT_LENGTH + } + + @pytest.mark.asyncio async def test_passthrough_success_reconciles_budget_reservation(): """ From e338a6b5ca72316595f518892195619b49cf16fa Mon Sep 17 00:00:00 2001 From: shrey kharbanda Date: Mon, 26 Oct 2026 14:52:42 -0700 Subject: [PATCH 03/10] refactor(passthrough): share the Deepgram WebSocket route body The listen route's body (provider credential, server-side DEEPGRAM_API_BASE, early refusal, price check, relay call) becomes _relay_deepgram_websocket, which takes a small _DeepgramWebsocketProduct record naming what differs per product: the target URL builder, the price check, the registry key and an optional refusal. The listen auth dependency and the listen route keep their names, paths and order In common_utils, the target URL builder and the upstream model lookup take the path and default model as arguments, and the exact registry lookup takes the registry key, so a second product can reuse them No behavior changes. The listen tests pass unchanged Refs LIT-10070 --- litellm/llms/deepgram/common_utils.py | 31 +++++++++--- .../llm_passthrough_endpoints.py | 49 ++++++++++--------- 2 files changed, 50 insertions(+), 30 deletions(-) diff --git a/litellm/llms/deepgram/common_utils.py b/litellm/llms/deepgram/common_utils.py index 29998895bbc..5638f2e0e4b 100644 --- a/litellm/llms/deepgram/common_utils.py +++ b/litellm/llms/deepgram/common_utils.py @@ -34,8 +34,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 +54,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,7 +101,10 @@ 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 _has_exact_registry_entry(deepgram_listen_registry_key(upstream_url)) + + +def _has_exact_registry_entry(registry_key: str) -> bool: try: model_info: Final = litellm.get_model_info(model=registry_key, custom_llm_provider=LlmProviders.DEEPGRAM.value) except Exception: diff --git a/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py b/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py index e478096f7ad..dfbe87bd496 100644 --- a/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py @@ -3469,12 +3469,21 @@ 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, ) -> None: deepgram_api_key: Final = passthrough_endpoint_router.get_credentials( custom_llm_provider=litellm.LlmProviders.DEEPGRAM.value, @@ -3485,23 +3494,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( @@ -3515,6 +3509,17 @@ async def deepgram_listen_websocket_route( ) +@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)) + + class BaseOpenAIPassThroughHandler: @staticmethod async def base_openai_pass_through_handler( From 930f4417246429c5ecbb64bf4b6913f0bd464958 Mon Sep 17 00:00:00 2001 From: shrey kharbanda Date: Mon, 26 Oct 2026 14:55:35 -0700 Subject: [PATCH 04/10] feat(passthrough): add Deepgram streaming text-to-speech WebSocket route Deepgram streams Aura text-to-speech over a WebSocket at /v1/speak. The proxy now relays it at /deepgram/v1/speak and /deepgram/speak with the same key auth (Bearer or a Deepgram SDK's Token scheme), provider key injection and server-configured DEEPGRAM_API_BASE as the listen route. A caller-supplied api_base is never honored The route names exactly one voice upstream, the first model in the query or aura-asteria-en when none is given, and authorizes the key against that same voice. A missing DEEPGRAM_API_KEY is closed with 1011 before the socket is accepted A session is billed at one cost-map row. An aura-2-- voice uses deepgram/aura-2, an aura-- voice uses deepgram/aura, and any other voice uses a row named after itself, deepgram/. When that row has no input_cost_per_character the socket is closed with 1008 before Deepgram is contacted, naming the row to add, so a new generation such as aura-3-x-en is refused rather than billed at another generation's rate, and adding its row is enough to enable it Spend is the relay's client text count times that rate. Only the length of each Speak message's text is counted, whether it arrives in a text or a binary frame, so Flush, Clear, Close and malformed frames add nothing, and the text never reaches the logging payload. The success handler gets one speak branch next to the listen branch, which takes its dispatcher one branch past the C901 limit, so the def carries a noqa with the reason That branch passes the dispatcher's untyped request_body and kwargs, as every branch does, so the listen channel_index check becomes a match statement to keep the type-check gate's unknown-argument count flat Refs LIT-10070 --- litellm/constants.py | 1 + litellm/llms/deepgram/common_utils.py | 77 ++++++- .../llm_passthrough_endpoints.py | 38 ++++ ...pgram_speak_passthrough_logging_handler.py | 79 +++++++ .../pass_through_endpoints/success_handler.py | 16 +- .../deepgram/test_deepgram_common_utils.py | 146 ++++++++++++ ...pgram_speak_passthrough_logging_handler.py | 174 ++++++++++++++ .../test_deepgram_ws_passthrough_routes.py | 214 +++++++++++++++++- .../test_pass_through_endpoints.py | 61 +++++ 9 files changed, 788 insertions(+), 18 deletions(-) create mode 100644 litellm/proxy/pass_through_endpoints/llm_provider_handlers/deepgram_speak_passthrough_logging_handler.py create mode 100644 tests/unit/proxy/pass_through_endpoints/llm_provider_handlers/test_deepgram_speak_passthrough_logging_handler.py diff --git a/litellm/constants.py b/litellm/constants.py index 1d5c6404bfd..796e4456bbb 100644 --- a/litellm/constants.py +++ b/litellm/constants.py @@ -423,6 +423,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 5638f2e0e4b..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"}) @@ -101,15 +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.""" - return _has_exact_registry_entry(deepgram_listen_registry_key(upstream_url)) + return _exact_registry_entry(deepgram_listen_registry_key(upstream_url)) is not None -def _has_exact_registry_entry(registry_key: str) -> bool: +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, ...]: @@ -132,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: @@ -214,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/proxy/pass_through_endpoints/llm_passthrough_endpoints.py b/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py index dfbe87bd496..4e4b5de90f8 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 @@ -3354,6 +3359,7 @@ class _WebsocketRelay(Protocol): forward_headers: bool, endpoint: str, accept_websocket: bool, + count_client_text: Callable[[str | bytes], int] | None = None, ) -> None: ... @@ -3484,6 +3490,7 @@ async def _relay_deepgram_websocket( 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, @@ -3506,6 +3513,7 @@ async def _relay_deepgram_websocket( forward_headers=False, endpoint=websocket.url.path, accept_websocket=False, + count_client_text=count_client_text, ) @@ -3520,6 +3528,36 @@ async def deepgram_listen_websocket_route( 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) + ) + + +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, + ) + + class BaseOpenAIPassThroughHandler: @staticmethod async def base_openai_pass_through_handler( 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/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/tests/unit/llms/deepgram/test_deepgram_common_utils.py b/tests/unit/llms/deepgram/test_deepgram_common_utils.py index 7f480d3459f..c0ea885aebf 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" @@ -344,3 +352,141 @@ def test_aura_rows_bill_speech_per_character_at_the_published_rate( 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): + 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): + """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): + 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): + 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): + 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): + """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): + 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..96247df28f1 --- /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): + 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(): + 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(): + 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): + 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): + 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): + 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 +): + """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 404ed0f5800..69106acb7e1 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(): 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): """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,198 @@ 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, monkeypatch): + 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): + """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, expected_target, monkeypatch): + 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): + """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(): + 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, missing_key, removed_row, monkeypatch): + """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(): + 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, expected_target, monkeypatch): + """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 a797f8084b9..abb59bc3b69 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 @@ -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 @@ -7954,6 +7955,66 @@ async def test_websocket_passthrough_counts_client_text_when_the_upstream_ends_w } +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): + self.payloads.append(kwargs["standard_logging_object"]) + + +@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): + """Runs the relay with the speak counter and the real success handler, so the spend logger sees what production + writes for one Speak message""" + capturing_logger = _CapturingSpendLogger() + monkeypatch.setattr(litellm, "_async_success_callback", [capturing_logger]) + monkeypatch.setattr(litellm, "success_callback", []) + monkeypatch.setattr(litellm, "callbacks", []) + queued: list[Coroutine[object, object, None]] = [] # mutable-ok: the fake logging worker keeps each coroutine + upstream_ws = SpeakingUpstreamWebSocket() + + 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 asyncio.wait_for( + websocket_passthrough_request( + websocket=_speak_client(), + target=DEEPGRAM_SPEAK_TARGET, + custom_headers={"Authorization": "Token dg-provider-key"}, + user_api_key_dict=UserAPIKeyAuth(api_key="hashed-key", team_id="team-tts"), + forward_headers=False, + endpoint="/deepgram/v1/speak", + accept_websocket=False, + count_client_text=deepgram_speak_characters, + ), + timeout=RELAY_HANG_GUARD_SECONDS, + ) + for coroutine in queued: + await coroutine + + assert len(capturing_logger.payloads) == 1 + payload = capturing_logger.payloads[0] + assert payload["model"] == "aura-2-thalia-en" + assert payload["custom_llm_provider"] == "deepgram" + # Aura-2 pay as you go rate, $0.030 per 1,000 characters, from https://deepgram.com/pricing, checked 2026-10-26 + assert payload["response_cost"] == pytest.approx(len(SPOKEN_TEXT) * 0.030 / 1000) + assert SPOKEN_TEXT not in json.dumps(payload, default=str) + + @pytest.mark.asyncio async def test_passthrough_success_reconciles_budget_reservation(): """ From 101cda9a5e3685a3fde41fdd0ee1f5ab8e49983a Mon Sep 17 00:00:00 2001 From: shrey kharbanda Date: Mon, 26 Oct 2026 18:57:56 -0700 Subject: [PATCH 05/10] test(passthrough): type the Deepgram speak tests Every new test and helper now annotates its parameters and returns, and every new binding is Final, per AGENTS.md. The two CustomLogger overrides keep the base class's untyped signature. --- .../deepgram/test_deepgram_common_utils.py | 18 ++-- ...pgram_speak_passthrough_logging_handler.py | 12 +-- .../test_deepgram_ws_passthrough_routes.py | 30 ++++--- .../test_pass_through_endpoints.py | 82 ++++++++++--------- 4 files changed, 80 insertions(+), 62 deletions(-) diff --git a/tests/unit/llms/deepgram/test_deepgram_common_utils.py b/tests/unit/llms/deepgram/test_deepgram_common_utils.py index c0ea885aebf..e5a95400a76 100644 --- a/tests/unit/llms/deepgram/test_deepgram_common_utils.py +++ b/tests/unit/llms/deepgram/test_deepgram_common_utils.py @@ -344,7 +344,7 @@ def test_deepgram_listen_addon_pricing_models(upstream_url: str, expected: tuple ) 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 @@ -384,7 +384,7 @@ SPEAK_BASE: Final = "wss://api.deepgram.com/v1/speak" ), ], ) -def test_deepgram_speak_websocket_target(api_base: str | None, query_string: str, expected: str): +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 @@ -392,7 +392,7 @@ def test_deepgram_speak_websocket_target(api_base: str | None, query_string: str "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): +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) @@ -421,7 +421,9 @@ def test_the_authorized_speak_model_is_the_only_model_the_upstream_target_carrie 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): +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}" @@ -438,19 +440,19 @@ def test_deepgram_speak_registry_key_names_the_aura_generation_row_else_the_voic pytest.param("nova-3", False, id="a listen model has no speech row"), ], ) -def test_deepgram_speak_is_priced(model: str, expected: bool): +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): +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): +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( @@ -488,5 +490,5 @@ def test_a_cost_map_row_named_after_a_new_voice_prices_it_at_that_rate(monkeypat 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): +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 index 96247df28f1..6186d8d53fd 100644 --- 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 @@ -43,7 +43,7 @@ pytestmark: Final = pytest.mark.usefixtures("local_model_cost_map") 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): +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 @@ -70,7 +70,7 @@ def _handler_cost(upstream_url: str, characters: object) -> float | None: return response_cost -def test_handler_bills_spoken_characters_at_the_aura_2_rate_and_names_the_voice(): +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( @@ -91,7 +91,7 @@ def test_handler_bills_spoken_characters_at_the_aura_2_rate_and_names_the_voice( ) -def test_handler_bills_an_aura_1_voice_at_the_aura_1_rate(): +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) @@ -99,12 +99,12 @@ def test_handler_bills_an_aura_1_voice_at_the_aura_1_rate(): "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): +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): +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 @@ -126,7 +126,7 @@ class _CapturingLogger(CustomLogger): ) 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() 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 69106acb7e1..9d78946cf59 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 @@ -108,13 +108,13 @@ async def _serve(websocket: _FakeWebSocket, user_api_key_dict: UserAPIKeyAuth | return relay -def test_deepgram_websocket_routes_registered(): +def test_deepgram_websocket_routes_registered() -> None: ws_paths = {route.path for route in router.routes if isinstance(route, WebSocketRoute)} assert {*LISTEN_PATHS, *SPEAK_PATHS} <= ws_paths @pytest.mark.parametrize("path", (*LISTEN_PATHS, *SPEAK_PATHS)) -def test_deepgram_websocket_routes_are_lazily_loaded_mapped_pass_through_routes(path): +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") @@ -492,7 +492,9 @@ async def _serve_speak(websocket: _FakeWebSocket, user_api_key_dict: UserAPIKeyA @pytest.mark.parametrize("path", SPEAK_PATHS) -async def test_deepgram_speak_forwards_query_injects_only_provider_auth_and_counts_characters(path, monkeypatch): +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") @@ -516,7 +518,7 @@ async def test_deepgram_speak_forwards_query_injects_only_provider_auth_and_coun assert websocket.closed is None -async def test_deepgram_listen_does_not_count_client_text(monkeypatch): +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) @@ -543,7 +545,9 @@ async def test_deepgram_listen_does_not_count_client_text(monkeypatch): ), ], ) -async def test_deepgram_speak_names_exactly_one_model_upstream(query, expected_target, monkeypatch): +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"): @@ -552,7 +556,9 @@ async def test_deepgram_speak_names_exactly_one_model_upstream(query, expected_t 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): +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( @@ -567,7 +573,7 @@ async def test_deepgram_speak_honours_server_configured_api_base_and_ignores_the ] -async def test_deepgram_speak_closes_before_accepting_when_provider_credentials_missing(): +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): @@ -590,7 +596,9 @@ async def test_deepgram_speak_closes_before_accepting_when_provider_credentials_ 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, missing_key, removed_row, monkeypatch): +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) @@ -609,7 +617,7 @@ async def test_deepgram_speak_refuses_sessions_it_cannot_price(query, missing_ke assert "dg-provider-key" not in websocket.closed[1] -def test_deepgram_speak_rejects_connections_without_a_litellm_key(): +def test_deepgram_speak_rejects_connections_without_a_litellm_key() -> None: relay: Final = _FakeRelay() client: Final = TestClient(_app_with_relay(relay)) @@ -641,7 +649,9 @@ def test_deepgram_speak_rejects_connections_without_a_litellm_key(): pytest.param("model=&encoding=mulaw", None, id="voice blank, default denied"), ], ) -def test_deepgram_speak_authorizes_the_voice_it_will_actually_send_upstream(query, expected_target, monkeypatch): +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) 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 abb59bc3b69..de9db053909 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 @@ -7810,24 +7810,24 @@ async def _track_cost_for_passthrough_kwargs(kwargs: dict) -> AsyncMock: return increment_spend_counters -DEEPGRAM_SPEAK_TARGET = "wss://api.deepgram.com/v1/speak?model=aura-2-thalia-en" -SPEAK_AUDIO_CHUNK = bytes(range(256)) * 8 -SPOKEN_TEXT = "The quick brown fox jumps over the lazy dog." -SPEAK_FRAME = json.dumps({"type": "Speak", "text": SPOKEN_TEXT}) -FLUSH_FRAME = json.dumps({"type": "Flush"}) -FLUSHED_FRAME = json.dumps({"type": "Flushed", "sequence_id": 0}) -SPEAK_CLOSE_FRAME = json.dumps({"type": "Close"}) -CLIENT_BINARY_FRAME = b"\x00\xffclient-binary" -SPEAK_SESSION_FRAMES = ({"text": SPEAK_FRAME}, {"bytes": CLIENT_BINARY_FRAME}, {"text": FLUSH_FRAME}) -SPEAK_SESSION_CLIENT_LENGTH = sum(map(len, (SPEAK_FRAME, CLIENT_BINARY_FRAME, FLUSH_FRAME, SPEAK_CLOSE_FRAME))) -RELAY_HANG_GUARD_SECONDS = 5.0 +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))) +RELAY_HANG_GUARD_SECONDS: Final = 5.0 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 = ""): + def __init__(self, close_code: int = 1000, close_reason: str = "") -> None: self.received: list[str | bytes] = [] self.close = AsyncMock() self._close_code = close_code @@ -7848,36 +7848,40 @@ class SpeakingUpstreamWebSocket: from websockets.exceptions import ConnectionClosedError, ConnectionClosedOK from websockets.frames import Close - reply = await self._replies.get() + reply: Final = await self._replies.get() if reply is not None: return reply - close = Close(self._close_code, self._close_reason) - closed = ConnectionClosedOK if self._close_code == 1000 else ConnectionClosedError + 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): +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 = asyncio.Event() + first_audio: Final = asyncio.Event() - async def script(): + 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 = script() + steps: Final = script() - async def receive(): + async def receive() -> dict[str, str | bytes]: return await anext(steps) - websocket = _client_websocket(receive) + websocket: Final = _client_websocket(receive) websocket.send_bytes = AsyncMock(side_effect=lambda data: first_audio.set()) return websocket -async def _relay_deepgram_speak(upstream_ws, websocket, count_client_text): +async def _relay_deepgram_speak( + upstream_ws: SpeakingUpstreamWebSocket, + websocket: MagicMock, + count_client_text: Callable[[str | bytes], int] | None, +) -> AsyncMock: """Runs the generic relay the way the Deepgram speak route does and returns the success handler mock""" with ( _patched_websocket_passthrough_environment(upstream_ws), @@ -7904,13 +7908,13 @@ async def _relay_deepgram_speak(upstream_ws, websocket, count_client_text): @pytest.mark.asyncio -async def test_websocket_passthrough_counts_client_text_without_changing_what_is_relayed(): +async def test_websocket_passthrough_counts_client_text_without_changing_what_is_relayed() -> None: """Audio reaches the client while the session is open, every client frame reaches the upstream unchanged, and the success handler gets only the summed count of the text and binary frames, never the frames""" - upstream_ws = SpeakingUpstreamWebSocket() - websocket = _speak_client() + upstream_ws: Final = SpeakingUpstreamWebSocket() + websocket: Final = _speak_client() - success_handler = await _relay_deepgram_speak(upstream_ws, websocket, count_client_text=len) + success_handler: Final = await _relay_deepgram_speak(upstream_ws, websocket, 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] @@ -7922,11 +7926,11 @@ async def test_websocket_passthrough_counts_client_text_without_changing_what_is @pytest.mark.asyncio -async def test_websocket_passthrough_without_a_counter_reports_an_empty_request_body(): +async def test_websocket_passthrough_without_a_counter_reports_an_empty_request_body() -> None: """Listen, realtime and every other caller pass no counter, so their success handler sees what it saw before""" - upstream_ws = SpeakingUpstreamWebSocket() + upstream_ws: Final = SpeakingUpstreamWebSocket() - success_handler = await _relay_deepgram_speak(upstream_ws, _speak_client(), count_client_text=None) + success_handler: Final = await _relay_deepgram_speak(upstream_ws, _speak_client(), count_client_text=None) assert upstream_ws.received == [SPEAK_FRAME, CLIENT_BINARY_FRAME, FLUSH_FRAME, SPEAK_CLOSE_FRAME] assert success_handler.call_args.kwargs["request_body"] == {} @@ -7943,11 +7947,11 @@ async def test_websocket_passthrough_without_a_counter_reports_an_empty_request_ ) async def test_websocket_passthrough_counts_client_text_when_the_upstream_ends_with_an_error_close( close_code: int, close_reason: str -): - upstream_ws = SpeakingUpstreamWebSocket(close_code=close_code, close_reason=close_reason) - websocket = _speak_client() +) -> None: + upstream_ws: Final = SpeakingUpstreamWebSocket(close_code=close_code, close_reason=close_reason) + websocket: Final = _speak_client() - success_handler = await _relay_deepgram_speak(upstream_ws, websocket, count_client_text=len) + success_handler: Final = await _relay_deepgram_speak(upstream_ws, websocket, count_client_text=len) websocket.close.assert_awaited_once_with(code=close_code, reason=close_reason) assert success_handler.call_args.kwargs["request_body"] == { @@ -7966,15 +7970,17 @@ class _CapturingSpendLogger(CustomLogger): @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): +async def test_deepgram_speak_session_logs_character_spend_once_and_never_the_spoken_text( + monkeypatch: pytest.MonkeyPatch, +) -> None: """Runs the relay with the speak counter and the real success handler, so the spend logger sees what production writes for one Speak message""" - capturing_logger = _CapturingSpendLogger() + capturing_logger: Final = _CapturingSpendLogger() monkeypatch.setattr(litellm, "_async_success_callback", [capturing_logger]) monkeypatch.setattr(litellm, "success_callback", []) monkeypatch.setattr(litellm, "callbacks", []) - queued: list[Coroutine[object, object, None]] = [] # mutable-ok: the fake logging worker keeps each coroutine - upstream_ws = SpeakingUpstreamWebSocket() + queued: Final[list[Coroutine[object, object, None]]] = [] # mutable-ok: the fake worker keeps each coroutine + upstream_ws: Final = SpeakingUpstreamWebSocket() with ( patch("litellm.proxy.proxy_server.proxy_logging_obj") as proxy_logging, @@ -8007,7 +8013,7 @@ async def test_deepgram_speak_session_logs_character_spend_once_and_never_the_sp await coroutine assert len(capturing_logger.payloads) == 1 - payload = capturing_logger.payloads[0] + payload: Final = capturing_logger.payloads[0] assert payload["model"] == "aura-2-thalia-en" assert payload["custom_llm_provider"] == "deepgram" # Aura-2 pay as you go rate, $0.030 per 1,000 characters, from https://deepgram.com/pricing, checked 2026-10-26 From df41566be9949dae06cc2c63db9a30d3b76dc06b Mon Sep 17 00:00:00 2001 From: shrey kharbanda Date: Mon, 26 Oct 2026 19:05:55 -0700 Subject: [PATCH 06/10] test(passthrough): bill the speak relay tests through the real success handler The relay tests reuse the hang-up PR's helper, which runs the real success handler and returns what a spend logger receives, instead of mocking our own handler. tests/unit/AGENTS.md rules out doubles on our own code. The empty request body test is dropped, since the spend a session logs is now the assertion --- ...pgram_speak_passthrough_logging_handler.py | 2 +- .../test_pass_through_endpoints.py | 204 +++++++++--------- 2 files changed, 97 insertions(+), 109 deletions(-) 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 index 6186d8d53fd..82e0d388607 100644 --- 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 @@ -113,7 +113,7 @@ class _CapturingLogger(CustomLogger): 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): + async def async_log_success_event(self, kwargs, response_obj, start_time, end_time) -> None: self.payloads.append(kwargs["standard_logging_object"]) 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 de9db053909..797e743ef2c 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 @@ -67,10 +67,10 @@ from litellm.types.guardrails import GuardrailEventHooks from litellm.types.passthrough_endpoints.pass_through_endpoints import ( LITELLM_PASS_THROUGH_DEPLOYMENT_MODEL_INFO_STATE_KEY, LITELLM_PASS_THROUGH_RAW_BODY_STATE_KEY, - WEBSOCKET_CLIENT_TEXT_COUNT_KEY, 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 @@ -7717,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 @@ -7820,7 +7880,6 @@ 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))) -RELAY_HANG_GUARD_SECONDS: Final = 5.0 class SpeakingUpstreamWebSocket: @@ -7877,67 +7936,35 @@ def _speak_client(frames: tuple[dict[str, str | bytes], ...] = SPEAK_SESSION_FRA return websocket -async def _relay_deepgram_speak( - upstream_ws: SpeakingUpstreamWebSocket, - websocket: MagicMock, - count_client_text: Callable[[str | bytes], int] | None, -) -> AsyncMock: - """Runs the generic relay the way the Deepgram speak route does and returns the success handler mock""" - with ( - _patched_websocket_passthrough_environment(upstream_ws), - patch( # test-quality-ok: pass_through_endpoint_logging is a module global read inside websocket_passthrough_request; there is no injection seam - "litellm.proxy.pass_through_endpoints.pass_through_endpoints." - "pass_through_endpoint_logging.pass_through_async_success_handler", - new=AsyncMock(), - ) as success_handler, - ): - await asyncio.wait_for( - websocket_passthrough_request( - websocket=websocket, - target=DEEPGRAM_SPEAK_TARGET, - custom_headers={"Authorization": "Token dg-provider-key"}, - user_api_key_dict=UserAPIKeyAuth(), - forward_headers=False, - endpoint="/deepgram/v1/speak", - accept_websocket=False, - count_client_text=count_client_text, - ), - timeout=RELAY_HANG_GUARD_SECONDS, - ) - return success_handler +def _aura_2_usd_per_character() -> float: + return litellm.model_cost["deepgram/aura-2"]["input_cost_per_character"] @pytest.mark.asyncio -async def test_websocket_passthrough_counts_client_text_without_changing_what_is_relayed() -> None: +@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 success handler gets only the summed count of the text and binary frames, never the frames""" + the session is billed for the summed count of the text and binary frames""" upstream_ws: Final = SpeakingUpstreamWebSocket() websocket: Final = _speak_client() - success_handler: Final = await _relay_deepgram_speak(upstream_ws, websocket, count_client_text=len) + 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 success_handler.call_args.kwargs["request_body"] == { - WEBSOCKET_CLIENT_TEXT_COUNT_KEY: SPEAK_SESSION_CLIENT_LENGTH - } + 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 -async def test_websocket_passthrough_without_a_counter_reports_an_empty_request_body() -> None: - """Listen, realtime and every other caller pass no counter, so their success handler sees what it saw before""" - upstream_ws: Final = SpeakingUpstreamWebSocket() - - success_handler: Final = await _relay_deepgram_speak(upstream_ws, _speak_client(), count_client_text=None) - - assert upstream_ws.received == [SPEAK_FRAME, CLIENT_BINARY_FRAME, FLUSH_FRAME, SPEAK_CLOSE_FRAME] - assert success_handler.call_args.kwargs["request_body"] == {} - assert success_handler.call_args.kwargs["response_body"] == [json.loads(FLUSHED_FRAME)] - - -@pytest.mark.asyncio +@pytest.mark.usefixtures("local_model_cost_map") @pytest.mark.parametrize( ("close_code", "close_reason"), [ @@ -7946,26 +7973,19 @@ async def test_websocket_passthrough_without_a_counter_reports_an_empty_request_ ], ) async def test_websocket_passthrough_counts_client_text_when_the_upstream_ends_with_an_error_close( - close_code: int, close_reason: str + 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() - success_handler: Final = await _relay_deepgram_speak(upstream_ws, websocket, count_client_text=len) + 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 success_handler.call_args.kwargs["request_body"] == { - WEBSOCKET_CLIENT_TEXT_COUNT_KEY: SPEAK_SESSION_CLIENT_LENGTH - } - - -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): - self.payloads.append(kwargs["standard_logging_object"]) + assert [log["response_cost"] for log in session.spend_logs] == [ + pytest.approx(SPEAK_SESSION_CLIENT_LENGTH * _aura_2_usd_per_character()) + ] @pytest.mark.asyncio @@ -7973,52 +7993,20 @@ class _CapturingSpendLogger(CustomLogger): async def test_deepgram_speak_session_logs_character_spend_once_and_never_the_spoken_text( monkeypatch: pytest.MonkeyPatch, ) -> None: - """Runs the relay with the speak counter and the real success handler, so the spend logger sees what production - writes for one Speak message""" - 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 - upstream_ws: Final = SpeakingUpstreamWebSocket() + session: Final = await _relay_and_capture_spend( + monkeypatch, + SpeakingUpstreamWebSocket(), + _speak_client(), + DEEPGRAM_SPEAK_TARGET, + "/deepgram/v1/speak", + count_client_text=deepgram_speak_characters, + ) - 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 asyncio.wait_for( - websocket_passthrough_request( - websocket=_speak_client(), - target=DEEPGRAM_SPEAK_TARGET, - custom_headers={"Authorization": "Token dg-provider-key"}, - user_api_key_dict=UserAPIKeyAuth(api_key="hashed-key", team_id="team-tts"), - forward_headers=False, - endpoint="/deepgram/v1/speak", - accept_websocket=False, - count_client_text=deepgram_speak_characters, - ), - timeout=RELAY_HANG_GUARD_SECONDS, - ) - for coroutine in queued: - await coroutine - - 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" - # Aura-2 pay as you go rate, $0.030 per 1,000 characters, from https://deepgram.com/pricing, checked 2026-10-26 - assert payload["response_cost"] == pytest.approx(len(SPOKEN_TEXT) * 0.030 / 1000) - assert SPOKEN_TEXT not in json.dumps(payload, default=str) + 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 From 65c272337b550256cf9a9578cbfdd8014023e479 Mon Sep 17 00:00:00 2001 From: shrey kharbanda Date: Mon, 26 Oct 2026 19:40:37 -0700 Subject: [PATCH 07/10] fix(passthrough): bill a client frame whose upstream send is cancelled When the upstream ends the session while the relay is still inside a client frame's send, the task is cancelled after the frame may already be on the wire. Each send now goes through one helper that adds the frame's count on success and on cancellation, but not when the send itself fails --- .../pass_through_endpoints.py | 19 +++++++----- .../test_pass_through_endpoints.py | 29 +++++++++++++++++++ 2 files changed, 41 insertions(+), 7 deletions(-) diff --git a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py index 09652dfca2a..cb7197eed34 100644 --- a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py @@ -2641,9 +2641,18 @@ 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""" - nonlocal client_text_count try: while True: message = await websocket.receive() @@ -2712,13 +2721,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)) - if count_client_text is not None: - client_text_count += count_client_text(text_data) + 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) - if count_client_text is not None: - client_text_count += count_client_text(bytes_data) + await send_upstream(bytes_data) except asyncio.CancelledError: raise except Exception: 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 797e743ef2c..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 @@ -7988,6 +7988,35 @@ async def test_websocket_passthrough_counts_client_text_when_the_upstream_ends_w ] +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( From d6b3d4f486994f4162a65f5922fdd298f25ceefa Mon Sep 17 00:00:00 2001 From: shrey kharbanda Date: Wed, 28 Oct 2026 16:03:52 -0700 Subject: [PATCH 08/10] fix(deepgram): match only list channel_index in Results frames, as before the match rewrite --- litellm/llms/deepgram/common_utils.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/litellm/llms/deepgram/common_utils.py b/litellm/llms/deepgram/common_utils.py index a4a413370f7..f7bcdafdfc4 100644 --- a/litellm/llms/deepgram/common_utils.py +++ b/litellm/llms/deepgram/common_utils.py @@ -140,7 +140,7 @@ def _channel_count(value: object) -> int | None: def _results_channel_count(frame: Mapping[str, object]) -> int | None: match frame.get("channel_index"): - case [object(), int() as channels]: + case list([object(), object() as channels]): return _channel_count(channels) case _: return None From 527f48755ba6a6dae014c7ef2c99a9ba95487f0d Mon Sep 17 00:00:00 2001 From: shrey kharbanda Date: Wed, 28 Oct 2026 16:21:09 -0700 Subject: [PATCH 09/10] fix(cost-map): drop mode from the Aura rows so Deepgram isn't listed as supporting /v1/audio/speech --- litellm/model_prices_and_context_window_backup.json | 2 -- model_prices_and_context_window.json | 2 -- 2 files changed, 4 deletions(-) diff --git a/litellm/model_prices_and_context_window_backup.json b/litellm/model_prices_and_context_window_backup.json index 446af103664..6ce41af0eaa 100644 --- a/litellm/model_prices_and_context_window_backup.json +++ b/litellm/model_prices_and_context_window_backup.json @@ -22782,7 +22782,6 @@ "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" @@ -22795,7 +22794,6 @@ "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" diff --git a/model_prices_and_context_window.json b/model_prices_and_context_window.json index 446af103664..6ce41af0eaa 100644 --- a/model_prices_and_context_window.json +++ b/model_prices_and_context_window.json @@ -22782,7 +22782,6 @@ "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" @@ -22795,7 +22794,6 @@ "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" From 554d3bebbb7925f2ec04d58b063cebd5083afa8c Mon Sep 17 00:00:00 2001 From: shrey kharbanda Date: Wed, 28 Oct 2026 17:36:20 -0700 Subject: [PATCH 10/10] refactor(passthrough): trim the Deepgram speak helpers to what the route needs Inline the one-line speak wrappers, hand the character count to billing on the logging record instead of request_body, and drop the two log lines and the type-gate offsets in listen code. Remove tests that repeat a check another layer already makes. --- litellm/constants.py | 1 + litellm/llms/deepgram/common_utils.py | 39 ++---- .../llm_passthrough_endpoints.py | 15 ++- ...pgram_speak_passthrough_logging_handler.py | 68 ++-------- .../pass_through_endpoints.py | 13 +- .../pass_through_endpoints/success_handler.py | 9 +- .../deepgram/test_deepgram_common_utils.py | 52 ++------ ...pgram_speak_passthrough_logging_handler.py | 123 +++--------------- .../test_deepgram_ws_passthrough_routes.py | 28 ---- 9 files changed, 72 insertions(+), 276 deletions(-) diff --git a/litellm/constants.py b/litellm/constants.py index 796e4456bbb..a6c73dc0eec 100644 --- a/litellm/constants.py +++ b/litellm/constants.py @@ -423,6 +423,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" +# 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" diff --git a/litellm/llms/deepgram/common_utils.py b/litellm/llms/deepgram/common_utils.py index f7bcdafdfc4..0b8f1d7f969 100644 --- a/litellm/llms/deepgram/common_utils.py +++ b/litellm/llms/deepgram/common_utils.py @@ -41,12 +41,12 @@ class DeepgramException(BaseLLMException): pass -def _requested_model(query_string: str, default_model: str) -> str: +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 _requested_model(query_string, DEEPGRAM_LISTEN_DEFAULT_MODEL) + return deepgram_requested_model(query_string, DEEPGRAM_LISTEN_DEFAULT_MODEL) def _first_occurrences(query_string: str) -> httpx.QueryParams: @@ -61,7 +61,7 @@ def _first_occurrences(query_string: str) -> httpx.QueryParams: ) -def _deepgram_websocket_target(api_base: str | None, path: str, query_string: str, default_model: str) -> str: +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) @@ -70,20 +70,20 @@ def _deepgram_websocket_target(api_base: str | None, path: str, query_string: st 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) + 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 _upstream_model(upstream_url: str, default_model: 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 default_model def deepgram_listen_model(upstream_url: str) -> str: - return _upstream_model(upstream_url, DEEPGRAM_LISTEN_DEFAULT_MODEL) + return deepgram_upstream_model(upstream_url, DEEPGRAM_LISTEN_DEFAULT_MODEL) def _param_enabled(values: Sequence[str]) -> bool: @@ -139,11 +139,10 @@ def _channel_count(value: object) -> int | None: def _results_channel_count(frame: Mapping[str, object]) -> int | None: - match frame.get("channel_index"): - case list([object(), object() as channels]): - return _channel_count(channels) - case _: - return 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]) def _declared_channel_count(upstream_url: str) -> int | None: @@ -224,21 +223,9 @@ def deepgram_listen_transcript(websocket_messages: Sequence[Mapping[str, object] ) -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) + 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 ) @@ -250,10 +237,6 @@ def deepgram_speak_character_rate(upstream_url: str) -> float | None: 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) diff --git a/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py b/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py index 4e4b5de90f8..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,11 +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_is_priced, deepgram_speak_registry_key, - deepgram_speak_requested_model, - deepgram_speak_websocket_target, + 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 @@ -3530,12 +3531,12 @@ async def deepgram_listen_websocket_route( 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) + websocket, model=deepgram_requested_model(websocket.url.query, DEEPGRAM_SPEAK_DEFAULT_MODEL) ) def _deepgram_speak_refusal(target: str) -> str | None: - if deepgram_speak_is_priced(target): + if deepgram_speak_character_rate(target) is not None: return None return _DEEPGRAM_WS_UNPRICED_REASON.format(registry_key=deepgram_speak_registry_key(target)) @@ -3547,7 +3548,9 @@ async def deepgram_speak_websocket_route( 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) + 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, 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 index 572d183ecc2..0473c527a2a 100644 --- 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 @@ -1,39 +1,14 @@ 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.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_speak_model, - deepgram_speak_registry_key, -) -from litellm.proxy._types import PassThroughEndpointLoggingTypedDict +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 -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 @@ -43,37 +18,18 @@ class DeepgramSpeakPassthroughLoggingHandler: @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) + return request_method == "WEBSOCKET" and "/deepgram/" in path and path.endswith("/speak") @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: 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"] = 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 + 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 cb7197eed34..4df2b20b89f 100644 --- a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py @@ -2490,7 +2490,7 @@ def _with_trace_context(headers: Mapping[str, str], parent_span: object) -> dict async def websocket_passthrough_request( websocket: WebSocket, target: str, - custom_headers: Mapping[str, str], + custom_headers: dict, user_api_key_dict: UserAPIKeyAuth, forward_headers: bool | None = False, endpoint: str | None = None, @@ -2512,8 +2512,9 @@ async def websocket_passthrough_request( 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 + 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 @@ -2849,6 +2850,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( @@ -2860,9 +2863,7 @@ async def websocket_passthrough_request( end_time=end_time, logging_obj=logging_obj, cache_hit=False, - request_body={} - if count_client_text is None - else {WEBSOCKET_CLIENT_TEXT_COUNT_KEY: client_text_count}, + request_body={}, **success_kwargs, ) ) diff --git a/litellm/proxy/pass_through_endpoints/success_handler.py b/litellm/proxy/pass_through_endpoints/success_handler.py index 8ad1a221f52..c7a7d0d0808 100644 --- a/litellm/proxy/pass_through_endpoints/success_handler.py +++ b/litellm/proxy/pass_through_endpoints/success_handler.py @@ -404,16 +404,11 @@ 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 = ( + standard_logging_response_object = ( # rebind-ok: elif-chain DeepgramSpeakPassthroughLoggingHandler.deepgram_speak_passthrough_handler( - request_body=request_body, - logging_obj=logging_obj, - upstream_url=str(httpx_response.request.url), - kwargs=kwargs, + logging_obj=logging_obj, upstream_url=str(httpx_response.request.url) ) ) - 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/tests/unit/llms/deepgram/test_deepgram_common_utils.py b/tests/unit/llms/deepgram/test_deepgram_common_utils.py index e5a95400a76..45f4722ecd8 100644 --- a/tests/unit/llms/deepgram/test_deepgram_common_utils.py +++ b/tests/unit/llms/deepgram/test_deepgram_common_utils.py @@ -7,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, @@ -19,13 +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_is_priced, - deepgram_speak_model, deepgram_speak_registry_key, - deepgram_speak_requested_model, - deepgram_speak_websocket_target, + deepgram_upstream_model, + deepgram_websocket_target, ) NOVA_3_URL: Final = "wss://api.deepgram.com/v1/listen?model=nova-3&encoding=linear16&sample_rate=16000" @@ -385,7 +385,7 @@ SPEAK_BASE: Final = "wss://api.deepgram.com/v1/speak" ], ) 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 + assert deepgram_websocket_target(api_base, "speak", query_string, DEEPGRAM_SPEAK_DEFAULT_MODEL) == expected @pytest.mark.parametrize( @@ -393,11 +393,12 @@ def test_deepgram_speak_websocket_target(api_base: str | None, query_string: str ["", "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) + """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"] == [deepgram_speak_requested_model(query_string)] - assert deepgram_speak_model(target) == deepgram_speak_requested_model(query_string) + assert parse_qs(urlparse(target).query)["model"] == [requested] + assert deepgram_upstream_model(target, DEEPGRAM_SPEAK_DEFAULT_MODEL) == requested @pytest.mark.parametrize( @@ -405,12 +406,6 @@ def test_the_authorized_speak_model_is_the_only_model_the_upstream_target_carrie [ # 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"), @@ -424,31 +419,7 @@ def test_the_authorized_speak_model_is_the_only_model_the_upstream_target_carrie 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 + assert deepgram_speak_registry_key(f"{SPEAK_BASE}?model={model}") == f"deepgram/{expected_family}" @pytest.mark.usefixtures("local_model_cost_map") @@ -462,7 +433,6 @@ def test_a_cost_map_row_named_after_a_new_voice_prices_it_at_that_rate(monkeypat ) assert deepgram_speak_character_rate(url) == 4e-05 - assert deepgram_speak_is_priced(url) is True @pytest.mark.parametrize( 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 index 82e0d388607..b2c821cf9bf 100644 --- 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 @@ -2,23 +2,15 @@ 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 +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" @@ -47,52 +39,41 @@ def test_is_deepgram_speak_route(url_route: str, request_method: str, expected: assert DeepgramSpeakPassthroughLoggingHandler.is_deepgram_speak_route(url_route, request_method) is expected -def _logging_obj(call_id: str = "call-dg-speak") -> LiteLLMLoggingObj: - return LiteLLMLoggingObj( +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_id, + 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 _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 _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_names_the_voice() -> None: - logging_obj: Final = _logging_obj() +def test_handler_bills_spoken_characters_at_the_aura_2_rate_and_logs_only_the_count() -> None: + logging_obj: Final = _logging_obj(SPOKEN_CHARACTERS) - 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": {}}}, - ) + result: Final = DeepgramSpeakPassthroughLoggingHandler.deepgram_speak_passthrough_handler(logging_obj, AURA_2_URL) - 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 ) + 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 _handler_cost(AURA_1_URL, 1000) == pytest.approx(1000 * AURA_1_USD_PER_CHARACTER) + assert _billed_cost(AURA_1_URL, 1000) == pytest.approx(1000 * AURA_1_USD_PER_CHARACTER) @pytest.mark.parametrize( @@ -100,75 +81,9 @@ def test_handler_bills_an_aura_1_voice_at_the_aura_1_rate() -> None: [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 + 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 _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" + 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 9d78946cf59..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 @@ -528,34 +528,6 @@ async def test_deepgram_listen_does_not_count_client_text(monkeypatch: pytest.Mo 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: