scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-06 07:48:06 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 112

refactor(passthrough): trim the Deepgram speak helpers to what the route needs
Some checks failed
LiteLLM Rust DB / db-queries (push) Has been cancelled
Details
LiteLLM Rust / rust-lint (push) Has been cancelled
Details
LiteLLM Rust / rust-test (push) Has been cancelled
Details
LiteLLM Rust / rust-wheel (push) Has been cancelled
Details

Browse source 
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.
This commit is contained in:
shrey kharbanda 2026-10-28 17:36:20 -07:00
parent 527f48755b
commit 554d3bebbb
9 changed files with 72 additions and 276 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
litellm/constants.py
View file

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

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

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

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

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

68
litellm/proxy/pass_through_endpoints/llm_provider_handlers/deepgram_speak_passthrough_logging_handler.py
View file

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

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

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

9
litellm/proxy/pass_through_endpoints/success_handler.py
View file

@ -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({}),

52
tests/unit/llms/deepgram/test_deepgram_common_utils.py
View file

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

123
tests/unit/proxy/pass_through_endpoints/llm_provider_handlers/test_deepgram_speak_passthrough_logging_handler.py
View file

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

28
tests/unit/proxy/pass_through_endpoints/test_deepgram_ws_passthrough_routes.py
View file

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