From 585c32d3f500d3788cf9a47928bc5922351d835b Mon Sep 17 00:00:00 2001 From: yassin Date: Thu, 17 Sep 2026 03:53:28 +0000 Subject: [PATCH] refactor(deepgram): move listen frame parsing into llms/deepgram and drop routine docstrings Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/llms/deepgram/common_utils.py | 63 +++++++++- .../llm_passthrough_endpoints.py | 8 -- ...gram_listen_passthrough_logging_handler.py | 71 +---------- .../pass_through_endpoints.py | 8 -- .../deepgram/test_deepgram_common_utils.py | 114 ++++++++++++++++++ ...gram_listen_passthrough_logging_handler.py | 51 -------- 6 files changed, 179 insertions(+), 136 deletions(-) create mode 100644 tests/test_litellm/llms/deepgram/test_deepgram_common_utils.py diff --git a/litellm/llms/deepgram/common_utils.py b/litellm/llms/deepgram/common_utils.py index db00d048f01..f1759f94775 100644 --- a/litellm/llms/deepgram/common_utils.py +++ b/litellm/llms/deepgram/common_utils.py @@ -1,5 +1,8 @@ +import math +from collections.abc import Mapping, Sequence from types import MappingProxyType from typing import Final +from urllib.parse import parse_qs, urlparse import httpx @@ -14,10 +17,6 @@ class DeepgramException(BaseLLMException): def deepgram_listen_websocket_target(api_base: str | None, query_string: str) -> str: - """ - The upstream ``/listen`` socket for a streaming transcription, keeping the client's query string as sent - and adding the default model only when the client named none - """ listen_url: Final = httpx.URL(f"{(api_base or DEEPGRAM_DEFAULT_API_BASE).rstrip('/')}/listen") websocket_url: Final = listen_url.copy_with(scheme=_WEBSOCKET_SCHEMES.get(listen_url.scheme, listen_url.scheme)) params: Final = httpx.QueryParams(query_string) @@ -25,3 +24,59 @@ def deepgram_listen_websocket_target(api_base: str | None, query_string: str) -> query_string if params.get("model") else str(params.remove("model").add("model", DEEPGRAM_LISTEN_DEFAULT_MODEL)) ) return f"{websocket_url}?{query}" + + +def deepgram_listen_model(upstream_url: str) -> str: + models: Final = parse_qs(urlparse(upstream_url).query).get("model") + return models[0] if models else DEEPGRAM_LISTEN_DEFAULT_MODEL + + +def _seconds(value: object) -> float | None: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + return float(value) if math.isfinite(value) and value >= 0 else None + + +def _results_frame_end(frame: Mapping[str, object]) -> float | None: + start: Final = _seconds(frame.get("start")) + duration: Final = _seconds(frame.get("duration")) + return None if start is None or duration is None else start + duration + + +def _final_transcript(frame: Mapping[str, object]) -> str | None: + if frame.get("is_final") is not True: + return None + channel: Final = frame.get("channel") + alternatives: Final = channel.get("alternatives") if isinstance(channel, Mapping) else None + first: Final = alternatives[0] if isinstance(alternatives, list) and alternatives else None + transcript: Final = first.get("transcript") if isinstance(first, Mapping) else None + return transcript if isinstance(transcript, str) and transcript else None + + +def deepgram_listen_audio_seconds(websocket_messages: Sequence[Mapping[str, object]]) -> float: + metadata_durations: Final = tuple( + duration + for frame in websocket_messages + if frame.get("type") == "Metadata" + if (duration := _seconds(frame.get("duration"))) is not None + ) + if metadata_durations: + return metadata_durations[-1] + return max( + ( + end + for frame in websocket_messages + if frame.get("type") == "Results" + if (end := _results_frame_end(frame)) is not None + ), + default=0.0, + ) + + +def deepgram_listen_transcript(websocket_messages: Sequence[Mapping[str, object]]) -> str: + return " ".join( + transcript + for frame in websocket_messages + if frame.get("type") == "Results" + if (transcript := _final_transcript(frame)) is not None + ) diff --git a/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py b/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py index a5a95c34910..1abbf90cb7a 100644 --- a/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py @@ -2613,10 +2613,6 @@ def _proxy_model_allowlists() -> _OpenAIWebsocketModelAllowlists: def _negotiated_websocket_subprotocol(websocket: WebSocket) -> str | None: - """ - The first subprotocol the client offered, echoed back so browsers that carry the LiteLLM key in - ``Sec-WebSocket-Protocol`` complete the handshake - """ requested_subprotocols: Final = tuple( protocol.strip() for protocol in (websocket.headers.get("sec-websocket-protocol") or "").split(",") @@ -2707,10 +2703,6 @@ async def deepgram_listen_websocket_route( user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth_websocket)], relay: Annotated[_WebsocketRelay, Depends(_websocket_relay)], ) -> None: - """ - Streaming speech to text through Deepgram's ``/v1/listen`` socket. Audio frames and transcript frames are - relayed unchanged; the call is billed on the audio duration Deepgram reports when the socket closes - """ deepgram_api_key: Final = passthrough_endpoint_router.get_credentials( custom_llm_provider=litellm.LlmProviders.DEEPGRAM.value, region_name=None, diff --git a/litellm/proxy/pass_through_endpoints/llm_provider_handlers/deepgram_listen_passthrough_logging_handler.py b/litellm/proxy/pass_through_endpoints/llm_provider_handlers/deepgram_listen_passthrough_logging_handler.py index 6a574a2e1b9..a5fea7c8020 100644 --- a/litellm/proxy/pass_through_endpoints/llm_provider_handlers/deepgram_listen_passthrough_logging_handler.py +++ b/litellm/proxy/pass_through_endpoints/llm_provider_handlers/deepgram_listen_passthrough_logging_handler.py @@ -1,81 +1,22 @@ -""" -Cost tracking for Deepgram's streaming ``/v1/listen`` WebSocket. Deepgram bills the audio it processed, which it -reports as ``duration`` on the closing ``Metadata`` frame; a stream that ends without one is billed on the furthest -``start + duration`` across its ``Results`` frames -""" - -import math from collections.abc import Mapping, Sequence from types import MappingProxyType from typing import Final -from urllib.parse import parse_qs, urlparse +from urllib.parse import urlparse import litellm from litellm._logging import verbose_proxy_logger -from litellm.constants import DEEPGRAM_LISTEN_DEFAULT_MODEL from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj +from litellm.llms.deepgram.common_utils import ( + deepgram_listen_audio_seconds, + deepgram_listen_model, + deepgram_listen_transcript, +) from litellm.proxy._types import PassThroughEndpointLoggingTypedDict from litellm.types.utils import TranscriptionResponse DEEPGRAM_LISTEN_ROUTE_SUFFIX: Final = "/listen" -def _seconds(value: object) -> float | None: - if isinstance(value, bool) or not isinstance(value, (int, float)): - return None - return float(value) if math.isfinite(value) and value >= 0 else None - - -def _results_frame_end(frame: Mapping[str, object]) -> float | None: - start: Final = _seconds(frame.get("start")) - duration: Final = _seconds(frame.get("duration")) - return None if start is None or duration is None else start + duration - - -def _final_transcript(frame: Mapping[str, object]) -> str | None: - if frame.get("is_final") is not True: - return None - channel: Final = frame.get("channel") - alternatives: Final = channel.get("alternatives") if isinstance(channel, Mapping) else None - first: Final = alternatives[0] if isinstance(alternatives, list) and alternatives else None - transcript: Final = first.get("transcript") if isinstance(first, Mapping) else None - return transcript if isinstance(transcript, str) and transcript else None - - -def deepgram_listen_audio_seconds(websocket_messages: Sequence[Mapping[str, object]]) -> float: - metadata_durations: Final = tuple( - duration - for frame in websocket_messages - if frame.get("type") == "Metadata" - if (duration := _seconds(frame.get("duration"))) is not None - ) - if metadata_durations: - return metadata_durations[-1] - return max( - ( - end - for frame in websocket_messages - if frame.get("type") == "Results" - if (end := _results_frame_end(frame)) is not None - ), - default=0.0, - ) - - -def deepgram_listen_transcript(websocket_messages: Sequence[Mapping[str, object]]) -> str: - return " ".join( - transcript - for frame in websocket_messages - if frame.get("type") == "Results" - if (transcript := _final_transcript(frame)) is not None - ) - - -def deepgram_listen_model(upstream_url: str) -> str: - models: Final = parse_qs(urlparse(upstream_url).query).get("model") - return models[0] if models else DEEPGRAM_LISTEN_DEFAULT_MODEL - - def _audio_cost(response: TranscriptionResponse, model: str) -> float | None: try: return litellm.completion_cost( diff --git a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py index cf6985852f2..449ae48b0ef 100644 --- a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py @@ -2121,9 +2121,6 @@ def _resolved_vertex_live_setup( def _json_object_frame(frame: str | bytes) -> dict[str, object] | None: - """ - The frame as a JSON object when it is one, for cost tracking; audio and non-object frames yield None - """ try: decoded: Final = json.loads(frame if isinstance(frame, str) else frame.decode("utf-8")) except (json.JSONDecodeError, UnicodeDecodeError): @@ -2431,10 +2428,6 @@ async def websocket_passthrough_request( json_frame_ordinal: Final = count() async def relay_upstream_frame(upstream_message: str | bytes) -> None: - """ - Send the frame to the client exactly as received, then keep it for cost tracking when it is a JSON - object; the Vertex AI Live setup acknowledgement only names the model, so it is read instead of kept - """ if isinstance(upstream_message, bytes): await websocket.send_bytes(upstream_message) else: @@ -2448,7 +2441,6 @@ async def websocket_passthrough_request( websocket_messages.append(message_data) async def forward_upstream_to_client() -> Close | None: - """Relay upstream frames to the client until the upstream closes, returning its close frame""" try: while True: await relay_upstream_frame(await upstream_ws.recv()) diff --git a/tests/test_litellm/llms/deepgram/test_deepgram_common_utils.py b/tests/test_litellm/llms/deepgram/test_deepgram_common_utils.py new file mode 100644 index 00000000000..a86cb83d628 --- /dev/null +++ b/tests/test_litellm/llms/deepgram/test_deepgram_common_utils.py @@ -0,0 +1,114 @@ +import math +from collections.abc import Mapping, Sequence +from typing import Final + +import pytest + +import litellm +from litellm.llms.deepgram.common_utils import ( + deepgram_listen_audio_seconds, + deepgram_listen_model, + deepgram_listen_transcript, + deepgram_listen_websocket_target, +) + +NOVA_3_URL: Final = "wss://api.deepgram.com/v1/listen?model=nova-3&encoding=linear16&sample_rate=16000" + + +def _results(start: object, duration: object, transcript: str = "", is_final: object = True) -> dict[str, object]: + return { + "type": "Results", + "start": start, + "duration": duration, + "is_final": is_final, + "channel": {"alternatives": [{"transcript": transcript, "confidence": 0.9}]}, + } + + +def _metadata(duration: object) -> dict[str, object]: + return {"type": "Metadata", "request_id": "req-1", "duration": duration, "channels": 1} + + +@pytest.mark.parametrize( + ("api_base", "query_string", "expected"), + [ + pytest.param( + None, + "model=nova-3&encoding=linear16", + "wss://api.deepgram.com/v1/listen?model=nova-3&encoding=linear16", + id="default", + ), + pytest.param( + None, + "encoding=linear16&sample_rate=16000", + "wss://api.deepgram.com/v1/listen?encoding=linear16&sample_rate=16000&model=nova-3", + id="model added when missing", + ), + pytest.param( + None, + "model=&encoding=linear16", + "wss://api.deepgram.com/v1/listen?encoding=linear16&model=nova-3", + id="empty model replaced", + ), + pytest.param( + "http://localhost:9000/v1/", + "model=nova-2", + "ws://localhost:9000/v1/listen?model=nova-2", + id="custom base becomes ws", + ), + pytest.param( + "wss://dg.internal/v1", + "model=nova-3&keywords=a&keywords=b", + "wss://dg.internal/v1/listen?model=nova-3&keywords=a&keywords=b", + id="repeated keys preserved", + ), + ], +) +def test_deepgram_listen_websocket_target(api_base: str | None, query_string: str, expected: str): + assert deepgram_listen_websocket_target(api_base=api_base, query_string=query_string) == expected + + +@pytest.mark.parametrize( + ("frames", "expected_seconds"), + [ + pytest.param((_results(0.0, 2.0), _results(2.0, 3.5), _metadata(6.25)), 6.25, id="metadata wins"), + pytest.param((_metadata(4.0), _results(0.0, 9.0), _metadata(5.5)), 5.5, id="last metadata wins"), + pytest.param((_results(0.0, 2.0), _results(2.0, 3.5), _results(1.0, 1.0)), 5.5, id="furthest results end"), + pytest.param((_results(0.0, 0.0), _metadata(0.0)), 0.0, id="zero metadata is a real zero"), + pytest.param((_results(0.0, 1.5), _metadata("6.25")), 1.5, id="string metadata is ignored"), + pytest.param((_results(0.0, 1.5), _metadata(True)), 1.5, id="boolean metadata is ignored"), + pytest.param((_results(0.0, 1.5), _metadata(-3.0)), 1.5, id="negative metadata is ignored"), + pytest.param((_results(0.0, 1.5), _metadata(math.nan), _metadata(math.inf)), 1.5, id="nan/inf ignored"), + pytest.param((_results("0", 2.0), _results(0.0, None), _results(0.0, 0.75)), 0.75, id="malformed results"), + pytest.param(({"type": "SpeechStarted", "timestamp": 3.0}, {"type": "UtteranceEnd"}), 0.0, id="no usage"), + pytest.param((), 0.0, id="no frames"), + ], +) +def test_deepgram_listen_audio_seconds(frames: Sequence[Mapping[str, object]], expected_seconds: float): + assert deepgram_listen_audio_seconds(frames) == expected_seconds + + +def test_deepgram_listen_transcript_joins_final_results_only(): + frames = ( + _results(0.0, 1.0, "hello wor", is_final=False), + _results(0.0, 1.5, "hello world"), + _results(1.5, 0.5, "", is_final=True), + _results(2.0, 1.0, "how are you", is_final="yes"), + {"type": "Results", "start": 3.0, "duration": 1.0, "is_final": True, "channel": {"alternatives": []}}, + _results(4.0, 1.0, "goodbye"), + _metadata(5.0), + ) + assert deepgram_listen_transcript(frames) == "hello world goodbye" + + +@pytest.mark.parametrize( + ("upstream_url", "expected_model"), + [ + (NOVA_3_URL, "nova-3"), + ("wss://api.deepgram.com/v1/listen?encoding=linear16&model=nova-2-medical", "nova-2-medical"), + ("wss://api.deepgram.com/v1/listen?model=nova-3&model=nova-2", "nova-3"), + ("wss://api.deepgram.com/v1/listen?encoding=linear16", litellm.constants.DEEPGRAM_LISTEN_DEFAULT_MODEL), + ], +) +def test_deepgram_listen_model_comes_from_the_upstream_query(upstream_url: str, expected_model: str): + assert deepgram_listen_model(upstream_url) == expected_model diff --git a/tests/test_litellm/proxy/pass_through_endpoints/llm_provider_handlers/test_deepgram_listen_passthrough_logging_handler.py b/tests/test_litellm/proxy/pass_through_endpoints/llm_provider_handlers/test_deepgram_listen_passthrough_logging_handler.py index f00411ac10c..40f520e344d 100644 --- a/tests/test_litellm/proxy/pass_through_endpoints/llm_provider_handlers/test_deepgram_listen_passthrough_logging_handler.py +++ b/tests/test_litellm/proxy/pass_through_endpoints/llm_provider_handlers/test_deepgram_listen_passthrough_logging_handler.py @@ -1,7 +1,5 @@ """Deepgram ``/v1/listen`` WebSocket passthrough: duration extraction and duration based cost tracking.""" -import math -from collections.abc import Mapping, Sequence from datetime import datetime from types import SimpleNamespace from typing import Final @@ -14,9 +12,6 @@ from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging from litellm.proxy._types import UserAPIKeyAuth from litellm.proxy.pass_through_endpoints.llm_provider_handlers.deepgram_listen_passthrough_logging_handler import ( DeepgramListenPassthroughLoggingHandler, - deepgram_listen_audio_seconds, - deepgram_listen_model, - deepgram_listen_transcript, ) from litellm.proxy.pass_through_endpoints.success_handler import PassThroughEndpointLogging from litellm.types.passthrough_endpoints.pass_through_endpoints import PassthroughStandardLoggingPayload @@ -39,52 +34,6 @@ def _metadata(duration: object) -> dict[str, object]: return {"type": "Metadata", "request_id": "req-1", "duration": duration, "channels": 1} -@pytest.mark.parametrize( - ("frames", "expected_seconds"), - [ - pytest.param((_results(0.0, 2.0), _results(2.0, 3.5), _metadata(6.25)), 6.25, id="metadata wins"), - pytest.param((_metadata(4.0), _results(0.0, 9.0), _metadata(5.5)), 5.5, id="last metadata wins"), - pytest.param((_results(0.0, 2.0), _results(2.0, 3.5), _results(1.0, 1.0)), 5.5, id="furthest results end"), - pytest.param((_results(0.0, 0.0), _metadata(0.0)), 0.0, id="zero metadata is a real zero"), - pytest.param((_results(0.0, 1.5), _metadata("6.25")), 1.5, id="string metadata is ignored"), - pytest.param((_results(0.0, 1.5), _metadata(True)), 1.5, id="boolean metadata is ignored"), - pytest.param((_results(0.0, 1.5), _metadata(-3.0)), 1.5, id="negative metadata is ignored"), - pytest.param((_results(0.0, 1.5), _metadata(math.nan), _metadata(math.inf)), 1.5, id="nan/inf ignored"), - pytest.param((_results("0", 2.0), _results(0.0, None), _results(0.0, 0.75)), 0.75, id="malformed results"), - pytest.param(({"type": "SpeechStarted", "timestamp": 3.0}, {"type": "UtteranceEnd"}), 0.0, id="no usage"), - pytest.param((), 0.0, id="no frames"), - ], -) -def test_deepgram_listen_audio_seconds(frames: Sequence[Mapping[str, object]], expected_seconds: float): - assert deepgram_listen_audio_seconds(frames) == expected_seconds - - -def test_deepgram_listen_transcript_joins_final_results_only(): - frames = ( - _results(0.0, 1.0, "hello wor", is_final=False), - _results(0.0, 1.5, "hello world"), - _results(1.5, 0.5, "", is_final=True), - _results(2.0, 1.0, "how are you", is_final="yes"), - {"type": "Results", "start": 3.0, "duration": 1.0, "is_final": True, "channel": {"alternatives": []}}, - _results(4.0, 1.0, "goodbye"), - _metadata(5.0), - ) - assert deepgram_listen_transcript(frames) == "hello world goodbye" - - -@pytest.mark.parametrize( - ("upstream_url", "expected_model"), - [ - (NOVA_3_URL, "nova-3"), - ("wss://api.deepgram.com/v1/listen?encoding=linear16&model=nova-2-medical", "nova-2-medical"), - ("wss://api.deepgram.com/v1/listen?model=nova-3&model=nova-2", "nova-3"), - ("wss://api.deepgram.com/v1/listen?encoding=linear16", litellm.constants.DEEPGRAM_LISTEN_DEFAULT_MODEL), - ], -) -def test_deepgram_listen_model_comes_from_the_upstream_query(upstream_url: str, expected_model: str): - assert deepgram_listen_model(upstream_url) == expected_model - - @pytest.mark.parametrize( ("url_route", "expected"), [