scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-23 00:41:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 81

refactor(deepgram): move listen frame parsing into llms/deepgram and drop routine docstrings

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yassin 2026-09-17 03:53:28 +00:00
parent 6e1b4959d1
commit 585c32d3f5
6 changed files with 179 additions and 136 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

63
litellm/llms/deepgram/common_utils.py
View file

@ -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
)

8
litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py
View file

@ -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,

71
litellm/proxy/pass_through_endpoints/llm_provider_handlers/deepgram_listen_passthrough_logging_handler.py
View file

@ -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(

8
litellm/proxy/pass_through_endpoints/pass_through_endpoints.py
View file

@ -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())

114
tests/test_litellm/llms/deepgram/test_deepgram_common_utils.py Normal file
View file

@ -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

51
tests/test_litellm/proxy/pass_through_endpoints/llm_provider_handlers/test_deepgram_listen_passthrough_logging_handler.py
View file

@ -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"),
[