diff --git a/litellm/constants.py b/litellm/constants.py index 338fe0f6b85..80055178be6 100644 --- a/litellm/constants.py +++ b/litellm/constants.py @@ -1571,6 +1571,10 @@ PASS_THROUGH_HEADER_PREFIX: Final = "x-pass-" BASE_MCP_ROUTE: Final = "/mcp" +TRANSCRIBE_JOB_POLLING_INTERVAL_SECONDS: Final = float(os.getenv("TRANSCRIBE_JOB_POLLING_INTERVAL_SECONDS", "10")) +TRANSCRIBE_JOB_MAX_POLLING_ATTEMPTS: Final = int(os.getenv("TRANSCRIBE_JOB_MAX_POLLING_ATTEMPTS", "720")) # 2 hours +TRANSCRIBE_MAX_MEDIA_DURATION_SECONDS: Final = 28800 # Amazon Transcribe quota: maximum audio file length + BATCH_STATUS_POLL_INTERVAL_SECONDS: Final = int(os.getenv("BATCH_STATUS_POLL_INTERVAL_SECONDS", 3600)) # 1 hour BATCH_STATUS_POLL_MAX_ATTEMPTS: Final = int(os.getenv("BATCH_STATUS_POLL_MAX_ATTEMPTS", 24)) # for 24 hours BATCH_TPD_WINDOW_SECONDS: Final = 86400 diff --git a/litellm/model_prices_and_context_window_backup.json b/litellm/model_prices_and_context_window_backup.json index e87a3fec99b..17aa7eea0c7 100644 --- a/litellm/model_prices_and_context_window_backup.json +++ b/litellm/model_prices_and_context_window_backup.json @@ -45747,6 +45747,16 @@ "/v1/audio/speech" ] }, + "transcribe/StartTranscriptionJob": { + "input_cost_per_second": 0.0001, + "litellm_provider": "transcribe", + "mode": "audio_transcription", + "output_cost_per_second": 0.0, + "source": "https://aws.amazon.com/transcribe/pricing/", + "metadata": { + "notes": "Amazon Transcribe standard batch transcription, billed per second of audio with no minimum. Same rate in every region of the AWS Price List offer file for transcribe (checked 2026-09-17)" + } + }, "aws_polly/standard": { "input_cost_per_character": 4e-06, "litellm_provider": "aws_polly", diff --git a/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py b/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py index 2a3dabcefdc..3115faca30f 100644 --- a/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py @@ -1338,7 +1338,9 @@ async def transcribe_proxy_route( from .llm_provider_handlers.transcribe_passthrough_logging_handler import ( TRANSCRIBE_CUSTOM_LLM_PROVIDER, TRANSCRIBE_TARGET_PREFIX, + transcribe_cost_per_second, transcribe_supported_operations, + transcribe_unpriceable_request_reason, ) if operation not in transcribe_supported_operations(): @@ -1366,6 +1368,9 @@ async def transcribe_proxy_route( raise HTTPException(status_code=400, detail="Request body must be a JSON object") if "stream" in data: raise HTTPException(status_code=400, detail="'stream' is not an Amazon Transcribe request member") + unpriceable_reason: Final = transcribe_unpriceable_request_reason(operation, data, transcribe_cost_per_second()) + if unpriceable_reason is not None: + raise HTTPException(status_code=400, detail=unpriceable_reason) from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM, run_aws_signing, sign_aws_json_post diff --git a/litellm/proxy/pass_through_endpoints/llm_provider_handlers/transcribe_passthrough_logging_handler.py b/litellm/proxy/pass_through_endpoints/llm_provider_handlers/transcribe_passthrough_logging_handler.py index 0cf593d28df..5a414b1febb 100644 --- a/litellm/proxy/pass_through_endpoints/llm_provider_handlers/transcribe_passthrough_logging_handler.py +++ b/litellm/proxy/pass_through_endpoints/llm_provider_handlers/transcribe_passthrough_logging_handler.py @@ -1,20 +1,104 @@ -from collections.abc import Mapping +import asyncio +import json +import math +from collections.abc import Awaitable, Callable, Mapping from datetime import datetime -from functools import lru_cache -from typing import Final +from functools import lru_cache, partial +from types import MappingProxyType +from typing import Final, Protocol, TypeAlias import httpx +from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError +from typing_extensions import ReadOnly, TypedDict +import litellm from litellm._logging import verbose_proxy_logger +from litellm.constants import ( + TRANSCRIBE_JOB_MAX_POLLING_ATTEMPTS, + TRANSCRIBE_JOB_POLLING_INTERVAL_SECONDS, + TRANSCRIBE_MAX_MEDIA_DURATION_SECONDS, +) +from litellm.litellm_core_utils.aws_partition import get_aws_dns_suffix from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.litellm_core_utils.litellm_logging import ( get_standard_logging_object_payload, ) -from litellm.proxy._types import PassThroughEndpointLoggingTypedDict +from litellm.llms.custom_httpx.http_handler import get_async_httpx_client +from litellm.proxy._types import PassThroughEndpointLoggingResultValues, PassThroughEndpointLoggingTypedDict +from litellm.types.llms.custom_http import httpxSpecialProvider from litellm.types.utils import StandardPassThroughResponseObject TRANSCRIBE_TARGET_PREFIX: Final = "Transcribe" TRANSCRIBE_CUSTOM_LLM_PROVIDER: Final = "transcribe" +TRANSCRIBE_PRICED_OPERATION: Final = "StartTranscriptionJob" +TRANSCRIBE_PRICED_MODEL: Final = f"{TRANSCRIBE_CUSTOM_LLM_PROVIDER}/{TRANSCRIBE_PRICED_OPERATION}" +TRANSCRIBE_UNPRICED_OPERATIONS: Final = frozenset( + {"StartCallAnalyticsJob", "StartMedicalScribeJob", "StartMedicalTranscriptionJob"} +) +TRANSCRIBE_SURCHARGE_MEMBERS: Final = ("ContentRedaction", "ToxicityDetection") +TRANSCRIBE_TERMINAL_JOB_STATUSES: Final = frozenset({"COMPLETED", "FAILED"}) + +JobLookup: TypeAlias = Callable[[str], Awaitable[Mapping[str, object]]] # mutable-ok: Callable parameter syntax +TranscriptFetch: TypeAlias = Callable[[str], Awaitable[Mapping[str, object]]] # mutable-ok: Callable parameter syntax +JobPricer: TypeAlias = Callable[[str, str, float], Awaitable[float]] # mutable-ok: Callable parameter syntax + + +class GetTranscriptionJobRequest(TypedDict): + TranscriptionJobName: ReadOnly[str] + + +class _TranscriptRef(BaseModel): + model_config = ConfigDict(frozen=True) + TranscriptFileUri: str | None = None + + +class _TranscriptionJob(BaseModel): + model_config = ConfigDict(frozen=True) + TranscriptionJobStatus: str | None = None + Transcript: _TranscriptRef | None = None + + +class _GetTranscriptionJobResponse(BaseModel): + model_config = ConfigDict(frozen=True) + TranscriptionJob: _TranscriptionJob | None = None + + +class _TranscriptItem(BaseModel): + model_config = ConfigDict(frozen=True) + end_time: float | None = None + + +class _TranscriptResults(BaseModel): + model_config = ConfigDict(frozen=True) + audio_segments: tuple[_TranscriptItem, ...] = () + items: tuple[_TranscriptItem, ...] = () + + +class _Transcript(BaseModel): + model_config = ConfigDict(frozen=True) + results: _TranscriptResults | None = None + + +class _PricedCostMapEntry(BaseModel): + model_config = ConfigDict(frozen=True, strict=True) + input_cost_per_second: float + + +_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object]) + + +class PassThroughLogDispatch(Protocol): + def __call__( + self, + *, + logging_obj: LiteLLMLoggingObj, + standard_logging_response_object: PassThroughEndpointLoggingResultValues | None, + result: str, + start_time: datetime, + end_time: datetime, + cache_hit: bool, + **kwargs: object, # kwargs-ok: mirrors the shared pass-through logging dispatch signature + ) -> Awaitable[None]: ... @lru_cache(maxsize=1) @@ -28,12 +112,265 @@ def transcribe_supported_operations() -> frozenset[str]: return frozenset(get_session().get_service_model("transcribe").operation_names) +def transcribe_cost_per_second() -> float | None: + try: + return _PricedCostMapEntry.model_validate(litellm.model_cost.get(TRANSCRIBE_PRICED_MODEL)).input_cost_per_second + except ValidationError: + return None + + +def transcribe_unpriceable_request_reason( + operation: str, + request_body: Mapping[str, object], + cost_per_second: float | None, +) -> str | None: + if operation in TRANSCRIBE_UNPRICED_OPERATIONS: + return ( + f"{operation} is billed per second of audio at a rate LiteLLM does not price yet, so it cannot be" + f" submitted through this route; only {TRANSCRIBE_PRICED_OPERATION} is priced and budgeted" + ) + if operation != TRANSCRIBE_PRICED_OPERATION: + return None + if cost_per_second is None: + return ( + f"{TRANSCRIBE_PRICED_MODEL} has no input_cost_per_second in the LiteLLM model cost map, so billable" + " transcription jobs cannot be submitted through this route" + ) + model_settings: Final = request_body.get("ModelSettings") + custom_language_model: Final = ( + ("ModelSettings.LanguageModelName",) + if isinstance(model_settings, Mapping) and "LanguageModelName" in model_settings + else () + ) + surcharges: Final = tuple(m for m in TRANSCRIBE_SURCHARGE_MEMBERS if m in request_body) + custom_language_model + if not surcharges: + return None + return ( + f"{TRANSCRIBE_PRICED_OPERATION} with {', '.join(surcharges)} adds a per-second surcharge LiteLLM does not" + " price yet; remove it to submit the job through this route" + ) + + +def transcription_job_cost(audio_seconds: float, cost_per_second: float) -> float: + return math.ceil(audio_seconds) * cost_per_second + + +def transcribe_max_job_cost(cost_per_second: float) -> float: + return transcription_job_cost(TRANSCRIBE_MAX_MEDIA_DURATION_SECONDS, cost_per_second) + + +def transcript_audio_seconds(transcript: Mapping[str, object]) -> float | None: + results: Final = _Transcript.model_validate(transcript).results + if results is None: + return None + end_times: Final = tuple( + item.end_time for item in results.audio_segments + results.items if item.end_time is not None + ) + return max(end_times, default=None) + + +async def await_transcription_job( + job_name: str, + get_job: JobLookup, + sleep: Callable[[float], Awaitable[None]] = asyncio.sleep, + max_attempts: int = TRANSCRIBE_JOB_MAX_POLLING_ATTEMPTS, +) -> _TranscriptionJob | None: + for _ in range(max_attempts): + job = _GetTranscriptionJobResponse.model_validate(await get_job(job_name)).TranscriptionJob + if job is not None and job.TranscriptionJobStatus in TRANSCRIBE_TERMINAL_JOB_STATUSES: + return job + await sleep(TRANSCRIBE_JOB_POLLING_INTERVAL_SECONDS) + return None + + +async def price_transcription_job( + job_name: str, + cost_per_second: float, + get_job: JobLookup, + fetch_transcript: TranscriptFetch, + sleep: Callable[[float], Awaitable[None]] = asyncio.sleep, + max_attempts: int = TRANSCRIBE_JOB_MAX_POLLING_ATTEMPTS, +) -> float: + """ + Amazon Transcribe bills per second of audio and reports the duration only inside the + transcript artifact, so the job is polled to completion and priced from the last end_time. + Anything that stops the duration from being read is charged as the longest media AWS accepts. + """ + job: Final = await await_transcription_job(job_name, get_job, sleep=sleep, max_attempts=max_attempts) + if job is None: + verbose_proxy_logger.warning("Transcribe job %s did not finish while polling, charging maximum", job_name) + return transcribe_max_job_cost(cost_per_second) + if job.TranscriptionJobStatus == "FAILED": + return 0.0 + transcript_uri: Final = job.Transcript.TranscriptFileUri if job.Transcript is not None else None + if transcript_uri is None: + return transcribe_max_job_cost(cost_per_second) + audio_seconds: Final = transcript_audio_seconds(await fetch_transcript(transcript_uri)) + if audio_seconds is None: + return transcribe_max_job_cost(cost_per_second) + return transcription_job_cost(audio_seconds, cost_per_second) + + +def _as_json_object(response: httpx.Response) -> Mapping[str, object]: + return _JSON_OBJECT.validate_python(response.raise_for_status().json()) + + +def transcribe_job_lookup(aws_region_name: str) -> JobLookup: + from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM, run_aws_signing, sign_aws_json_post + + url: Final = f"https://transcribe.{aws_region_name}.{get_aws_dns_suffix(aws_region_name)}/" + headers: Final = MappingProxyType( + { + "Content-Type": "application/x-amz-json-1.1", + "X-Amz-Target": f"{TRANSCRIBE_TARGET_PREFIX}.GetTranscriptionJob", + } + ) + + async def get_job(job_name: str) -> Mapping[str, object]: + body: Final[GetTranscriptionJobRequest] = {"TranscriptionJobName": job_name} + payload: Final = json.dumps(body) + prepped: Final = await run_aws_signing( + sign_aws_json_post, + get_credentials=partial(BaseAWSLLM().get_credentials, aws_region_name=aws_region_name), + service_name="transcribe", + aws_region_name=aws_region_name, + url=url, + body=payload, + headers=headers, + ) + client: Final = get_async_httpx_client(llm_provider=httpxSpecialProvider.PassThroughEndpoint) + signed_headers: Final = dict(prepped.headers.items()) # mutable-ok: AsyncHTTPHandler.post takes a dict + return _as_json_object(await client.post(str(prepped.url), data=payload, headers=signed_headers)) + + return get_job + + +def transcribe_transcript_fetch(aws_region_name: str) -> TranscriptFetch: + from botocore.auth import S3SigV4Auth + from botocore.awsrequest import AWSRequest + + from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM, run_aws_signing + + def sign_s3_get(transcript_uri: str) -> dict[str, str]: # mutable-ok: AsyncHTTPHandler.get takes a dict + aws_request: Final = AWSRequest(method="GET", url=transcript_uri) + credentials: Final = BaseAWSLLM().get_credentials(aws_region_name=aws_region_name) + S3SigV4Auth(credentials, "s3", aws_region_name).add_auth(aws_request) + return dict(aws_request.prepare().headers.items()) # mutable-ok: AsyncHTTPHandler.get takes a dict + + async def fetch_transcript(transcript_uri: str) -> Mapping[str, object]: + presigned: Final = "X-Amz-Signature" in httpx.URL(transcript_uri).params + headers: Final = None if presigned else await run_aws_signing(sign_s3_get, transcript_uri) + client: Final = get_async_httpx_client(llm_provider=httpxSpecialProvider.PassThroughEndpoint) + return _as_json_object(await client.get(transcript_uri, headers=headers)) + + return fetch_transcript + + +async def price_transcription_job_live(job_name: str, aws_region_name: str, cost_per_second: float) -> float: + try: + return await price_transcription_job( + job_name, + cost_per_second, + get_job=transcribe_job_lookup(aws_region_name), + fetch_transcript=transcribe_transcript_fetch(aws_region_name), + ) + except Exception as e: # noqa: BLE001 # an unreadable job must still be charged, so fail closed at the maximum + verbose_proxy_logger.exception("Pricing Transcribe job %s failed, charging maximum: %s", job_name, e) + return transcribe_max_job_cost(cost_per_second) + + class TranscribePassthroughLoggingHandler: + def __init__(self, job_pricer: JobPricer = price_transcription_job_live) -> None: + self._job_pricer: Final = job_pricer + self._pricing_tasks: Final[set[asyncio.Task[None]]] = set() # mutable-ok: asyncio holds tasks weakly + @staticmethod def _operation_from_response(httpx_response: httpx.Response) -> str: - target: Final = httpx_response.request.headers.get("x-amz-target", "") + headers: Final[Mapping[str, str]] = httpx_response.request.headers + target: Final = headers.get("x-amz-target", "") return target.split(".")[-1] + @staticmethod + def is_priced_job_start(httpx_response: httpx.Response) -> bool: + return ( + TranscribePassthroughLoggingHandler._operation_from_response(httpx_response) == TRANSCRIBE_PRICED_OPERATION + ) + + def schedule_priced_job_logging( + self, + httpx_response: httpx.Response, + logging_obj: LiteLLMLoggingObj, + url_route: str, + result: str, + start_time: datetime, + end_time: datetime, + cache_hit: bool, + request_body: Mapping[str, object], + log: PassThroughLogDispatch, + **kwargs: object, # kwargs-ok: the passthrough logging dispatch forwards shared logging kwargs to every handler + ) -> asyncio.Task[None]: + task: Final = asyncio.create_task( + self._price_then_log( + httpx_response=httpx_response, + logging_obj=logging_obj, + url_route=url_route, + result=result, + start_time=start_time, + end_time=end_time, + cache_hit=cache_hit, + request_body=request_body, + log=log, + **kwargs, + ) + ) + self._pricing_tasks.add(task) + task.add_done_callback(self._pricing_tasks.discard) + return task + + async def _price_then_log( + self, + httpx_response: httpx.Response, + logging_obj: LiteLLMLoggingObj, + url_route: str, + result: str, + start_time: datetime, + end_time: datetime, + cache_hit: bool, + request_body: Mapping[str, object], + log: PassThroughLogDispatch, + **kwargs: object, # kwargs-ok: the passthrough logging dispatch forwards shared logging kwargs to every handler + ) -> None: + cost_per_second: Final = transcribe_cost_per_second() + if cost_per_second is None: + verbose_proxy_logger.error("%s left the model cost map, spend not recorded", TRANSCRIBE_PRICED_MODEL) + return + job_name: Final = request_body.get("TranscriptionJobName") + aws_region_name: Final = httpx_response.request.url.host.split(".")[1] + response_cost: Final = await self._job_pricer( + job_name if isinstance(job_name, str) else "", aws_region_name, cost_per_second + ) + payload: Final = self.transcribe_passthrough_handler( + httpx_response=httpx_response, + logging_obj=logging_obj, + url_route=url_route, + result=result, + start_time=start_time, + end_time=end_time, + cache_hit=cache_hit, + request_body=request_body, + response_cost=response_cost, + **kwargs, + ) + await log( + logging_obj=logging_obj, + standard_logging_response_object=payload["result"], + result=result, + start_time=start_time, + end_time=end_time, + cache_hit=cache_hit, + **payload["kwargs"], + ) + @staticmethod def transcribe_passthrough_handler( httpx_response: httpx.Response, @@ -44,13 +381,9 @@ class TranscribePassthroughLoggingHandler: end_time: datetime, cache_hit: bool, request_body: Mapping[str, object], + response_cost: float = 0.0, **kwargs: object, # kwargs-ok: the passthrough logging dispatch forwards shared logging kwargs to every handler ) -> PassThroughEndpointLoggingTypedDict: - """ - Records model and provider for an Amazon Transcribe control-plane call. Transcribe - bills per second of audio once a job finishes, which no request or response on this - path carries, so response_cost is recorded as 0.0 rather than estimated. - """ try: operation: Final = TranscribePassthroughLoggingHandler._operation_from_response(httpx_response) model_name: Final = f"{TRANSCRIBE_CUSTOM_LLM_PROVIDER}/{operation}" @@ -59,12 +392,12 @@ class TranscribePassthroughLoggingHandler: **kwargs, "model": model_name, "custom_llm_provider": TRANSCRIBE_CUSTOM_LLM_PROVIDER, - "response_cost": 0.0, + "response_cost": response_cost, } logging_obj.model_call_details.update( model=model_name, custom_llm_provider=TRANSCRIBE_CUSTOM_LLM_PROVIDER, - response_cost=0.0, + response_cost=response_cost, ) standard_logging_object: Final = get_standard_logging_object_payload( diff --git a/litellm/proxy/pass_through_endpoints/success_handler.py b/litellm/proxy/pass_through_endpoints/success_handler.py index 7dfada592b8..43b9355e5b5 100644 --- a/litellm/proxy/pass_through_endpoints/success_handler.py +++ b/litellm/proxy/pass_through_endpoints/success_handler.py @@ -52,7 +52,10 @@ def _safe_response_text(httpx_response: httpx.Response) -> str: class PassThroughEndpointLogging: - def __init__(self): + def __init__(self, transcribe_handler: TranscribePassthroughLoggingHandler | None = None): + self.transcribe_passthrough_logging_handler: Final = ( + transcribe_handler if transcribe_handler is not None else TranscribePassthroughLoggingHandler() + ) self.TRACKED_VERTEX_METHOD_ROUTES = ( "generateContent", "streamGenerateContent", @@ -336,6 +339,23 @@ class PassThroughEndpointLogging: elif self.is_langfuse_route(url_route): # Don't log langfuse pass-through requests return + elif self.is_transcribe_route(custom_llm_provider) and TranscribePassthroughLoggingHandler.is_priced_job_start( + httpx_response + ): + self.transcribe_passthrough_logging_handler.schedule_priced_job_logging( + httpx_response=httpx_response, + logging_obj=logging_obj, + url_route=url_route, + result=result, + start_time=start_time, + end_time=end_time, + cache_hit=cache_hit, + request_body=request_body, + log=self._handle_logging, + standard_pass_through_logging_payload=passthrough_logging_payload, + **kwargs, + ) + return else: normalized_llm_passthrough_logging_payload: Final = self.normalize_llm_passthrough_logging_payload( httpx_response=httpx_response, diff --git a/model_prices_and_context_window.json b/model_prices_and_context_window.json index e87a3fec99b..17aa7eea0c7 100644 --- a/model_prices_and_context_window.json +++ b/model_prices_and_context_window.json @@ -45747,6 +45747,16 @@ "/v1/audio/speech" ] }, + "transcribe/StartTranscriptionJob": { + "input_cost_per_second": 0.0001, + "litellm_provider": "transcribe", + "mode": "audio_transcription", + "output_cost_per_second": 0.0, + "source": "https://aws.amazon.com/transcribe/pricing/", + "metadata": { + "notes": "Amazon Transcribe standard batch transcription, billed per second of audio with no minimum. Same rate in every region of the AWS Price List offer file for transcribe (checked 2026-09-17)" + } + }, "aws_polly/standard": { "input_cost_per_character": 4e-06, "litellm_provider": "aws_polly", diff --git a/tests/test_litellm/proxy/pass_through_endpoints/llm_provider_handlers/test_transcribe_passthrough_logging_handler.py b/tests/test_litellm/proxy/pass_through_endpoints/llm_provider_handlers/test_transcribe_passthrough_logging_handler.py index edaa0635da9..8fc17110d8f 100644 --- a/tests/test_litellm/proxy/pass_through_endpoints/llm_provider_handlers/test_transcribe_passthrough_logging_handler.py +++ b/tests/test_litellm/proxy/pass_through_endpoints/llm_provider_handlers/test_transcribe_passthrough_logging_handler.py @@ -1,16 +1,26 @@ +import asyncio from datetime import datetime from unittest.mock import MagicMock import httpx +import pytest +import litellm from litellm.proxy.pass_through_endpoints.llm_provider_handlers.transcribe_passthrough_logging_handler import ( + TRANSCRIBE_MAX_MEDIA_DURATION_SECONDS, TranscribePassthroughLoggingHandler, + price_transcription_job, + transcribe_cost_per_second, transcribe_supported_operations, + transcribe_unpriceable_request_reason, + transcript_audio_seconds, ) from litellm.proxy.pass_through_endpoints.success_handler import ( PassThroughEndpointLogging, ) +COST_PER_SECOND = 0.0001 + def _make_response(operation: str) -> httpx.Response: request = httpx.Request( @@ -28,6 +38,26 @@ def _make_logging_obj() -> MagicMock: return logging_obj +async def _no_sleep(_: float) -> None: + return None + + +def _job(status: str, transcript_uri: str | None = "https://s3.us-west-2.amazonaws.com/b/t.json") -> dict[str, object]: + transcript = {"Transcript": {"TranscriptFileUri": transcript_uri}} if transcript_uri else {} + return {"TranscriptionJob": {"TranscriptionJobStatus": status, **transcript}} + + +def _sequence(*jobs: dict[str, object]): + remaining = list(jobs) + seen: list[str] = [] + + async def get_job(job_name: str) -> dict[str, object]: + seen.append(job_name) + return remaining.pop(0) if len(remaining) > 1 else remaining[0] + + return get_job, seen + + class TestTranscribeSupportedOperations: def test_matches_the_installed_botocore_service_model(self): from botocore.session import get_session @@ -37,8 +67,142 @@ class TestTranscribeSupportedOperations: ) +class TestTranscribeCostMap: + def test_start_transcription_job_is_priced_per_second_of_audio(self): + entry = litellm.model_cost["transcribe/StartTranscriptionJob"] + + assert entry["litellm_provider"] == "transcribe" + assert entry["mode"] == "audio_transcription" + assert transcribe_cost_per_second() == entry["input_cost_per_second"] > 0 + + def test_missing_or_malformed_entry_yields_no_rate(self, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setitem(litellm.model_cost, "transcribe/StartTranscriptionJob", {"input_cost_per_second": "x"}) + assert transcribe_cost_per_second() is None + monkeypatch.delitem(litellm.model_cost, "transcribe/StartTranscriptionJob") + assert transcribe_cost_per_second() is None + + +class TestTranscribeUnpriceableRequestReason: + def test_plain_start_transcription_job_is_allowed(self): + body = {"TranscriptionJobName": "j", "Media": {"MediaFileUri": "s3://b/a.wav"}} + assert transcribe_unpriceable_request_reason("StartTranscriptionJob", body, COST_PER_SECOND) is None + + def test_read_only_operations_are_allowed_without_a_rate(self): + assert transcribe_unpriceable_request_reason("GetTranscriptionJob", {}, None) is None + assert transcribe_unpriceable_request_reason("ListTranscriptionJobs", {}, None) is None + + def test_start_transcription_job_needs_a_rate(self): + reason = transcribe_unpriceable_request_reason("StartTranscriptionJob", {"TranscriptionJobName": "j"}, None) + assert reason is not None and "model cost map" in reason + + @pytest.mark.parametrize( + "operation", ["StartCallAnalyticsJob", "StartMedicalScribeJob", "StartMedicalTranscriptionJob"] + ) + def test_unpriced_job_classes_are_rejected(self, operation: str): + reason = transcribe_unpriceable_request_reason(operation, {}, COST_PER_SECOND) + assert reason is not None and operation in reason + + @pytest.mark.parametrize( + ("body", "member"), + [ + ({"ContentRedaction": {"RedactionType": "PII", "RedactionOutput": "redacted"}}, "ContentRedaction"), + ({"ToxicityDetection": [{"ToxicityCategories": ["ALL"]}]}, "ToxicityDetection"), + ({"ModelSettings": {"LanguageModelName": "clm"}}, "ModelSettings.LanguageModelName"), + ], + ) + def test_surcharged_features_are_rejected(self, body: dict[str, object], member: str): + reason = transcribe_unpriceable_request_reason("StartTranscriptionJob", body, COST_PER_SECOND) + assert reason is not None and member in reason + + def test_model_settings_without_a_custom_model_is_allowed(self): + body = {"ModelSettings": {}} + assert transcribe_unpriceable_request_reason("StartTranscriptionJob", body, COST_PER_SECOND) is None + + +class TestTranscriptAudioSeconds: + def test_reads_the_last_segment_end_time(self): + transcript = { + "results": { + "audio_segments": [{"end_time": "9.5"}, {"end_time": "17.36"}], + "items": [{"end_time": "17.23"}, {"type": "punctuation"}], + } + } + assert transcript_audio_seconds(transcript) == 17.36 + + def test_falls_back_to_items_when_segments_are_absent(self): + assert transcript_audio_seconds({"results": {"items": [{"end_time": "3.1"}]}}) == 3.1 + + def test_without_timings_is_unknown(self): + assert transcript_audio_seconds({"results": {"items": []}}) is None + assert transcript_audio_seconds({"jobName": "j"}) is None + + +class TestPriceTranscriptionJob: + @pytest.mark.asyncio + async def test_polls_until_completed_then_charges_rounded_up_audio_seconds(self): + get_job, seen = _sequence(_job("IN_PROGRESS"), _job("IN_PROGRESS"), _job("COMPLETED")) + fetched: list[str] = [] + + async def fetch_transcript(uri: str) -> dict[str, object]: + fetched.append(uri) + return {"results": {"audio_segments": [{"end_time": "17.36"}]}} + + cost = await price_transcription_job("job-1", COST_PER_SECOND, get_job, fetch_transcript, sleep=_no_sleep) + + assert cost == pytest.approx(18 * COST_PER_SECOND) + assert seen == ["job-1", "job-1", "job-1"] + assert fetched == ["https://s3.us-west-2.amazonaws.com/b/t.json"] + + @pytest.mark.asyncio + async def test_failed_job_costs_nothing(self): + get_job, _ = _sequence(_job("FAILED", transcript_uri=None)) + + async def fetch_transcript(uri: str) -> dict[str, object]: + raise AssertionError("failed jobs have no transcript to fetch") + + assert ( + await price_transcription_job("job-1", COST_PER_SECOND, get_job, fetch_transcript, sleep=_no_sleep) == 0.0 + ) + + @pytest.mark.asyncio + async def test_job_that_never_finishes_is_charged_the_maximum(self): + get_job, seen = _sequence(_job("IN_PROGRESS")) + + async def fetch_transcript(uri: str) -> dict[str, object]: + raise AssertionError("unfinished jobs have no transcript to fetch") + + cost = await price_transcription_job( + "job-1", COST_PER_SECOND, get_job, fetch_transcript, sleep=_no_sleep, max_attempts=3 + ) + + assert cost == pytest.approx(TRANSCRIBE_MAX_MEDIA_DURATION_SECONDS * COST_PER_SECOND) + assert len(seen) == 3 + + @pytest.mark.asyncio + async def test_unreadable_transcript_is_charged_the_maximum(self): + get_job, _ = _sequence(_job("COMPLETED")) + + async def fetch_transcript(uri: str) -> dict[str, object]: + return {"results": {}} + + cost = await price_transcription_job("job-1", COST_PER_SECOND, get_job, fetch_transcript, sleep=_no_sleep) + + assert cost == pytest.approx(TRANSCRIBE_MAX_MEDIA_DURATION_SECONDS * COST_PER_SECOND) + + @pytest.mark.asyncio + async def test_completed_job_without_transcript_uri_is_charged_the_maximum(self): + get_job, _ = _sequence(_job("COMPLETED", transcript_uri=None)) + + async def fetch_transcript(uri: str) -> dict[str, object]: + raise AssertionError("no URI to fetch") + + cost = await price_transcription_job("job-1", COST_PER_SECOND, get_job, fetch_transcript, sleep=_no_sleep) + + assert cost == pytest.approx(TRANSCRIBE_MAX_MEDIA_DURATION_SECONDS * COST_PER_SECOND) + + class TestTranscribePassthroughHandler: - def test_records_model_provider_and_zero_cost(self): + def test_records_model_provider_and_the_given_cost(self): logging_obj = _make_logging_obj() request_body = {"TranscriptionJobName": "litellm-job-1"} @@ -51,18 +215,130 @@ class TestTranscribePassthroughHandler: end_time=datetime.now(), cache_hit=False, request_body=request_body, + response_cost=0.0018, ) assert handler_result["result"] == {"response": '{"TranscriptionJob": {}}'} assert handler_result["kwargs"]["model"] == "transcribe/StartTranscriptionJob" assert handler_result["kwargs"]["custom_llm_provider"] == "transcribe" - assert handler_result["kwargs"]["response_cost"] == 0.0 - assert "standard_logging_object" in handler_result["kwargs"] + assert handler_result["kwargs"]["response_cost"] == 0.0018 + assert handler_result["kwargs"]["standard_logging_object"]["response_cost"] == 0.0018 assert logging_obj.model_call_details["model"] == "transcribe/StartTranscriptionJob" assert logging_obj.model_call_details["custom_llm_provider"] == "transcribe" - assert logging_obj.model_call_details["response_cost"] == 0.0 + assert logging_obj.model_call_details["response_cost"] == 0.0018 assert request_body == {"TranscriptionJobName": "litellm-job-1"} + def test_read_only_operations_default_to_zero_cost(self): + handler_result = TranscribePassthroughLoggingHandler.transcribe_passthrough_handler( + httpx_response=_make_response("GetTranscriptionJob"), + logging_obj=_make_logging_obj(), + url_route="https://transcribe.us-west-2.amazonaws.com/", + result='{"TranscriptionJob": {}}', + start_time=datetime.now(), + end_time=datetime.now(), + cache_hit=False, + request_body={"TranscriptionJobName": "litellm-job-1"}, + ) + + assert handler_result["kwargs"]["response_cost"] == 0.0 + + +class TestStartTranscriptionJobIsLoggedAtJobCost: + @pytest.mark.asyncio + async def test_success_handler_defers_logging_until_the_job_is_priced(self): + priced: list[tuple[str, str, float]] = [] + + async def job_pricer(job_name: str, aws_region_name: str, cost_per_second: float) -> float: + priced.append((job_name, aws_region_name, cost_per_second)) + return 0.0018 + + logged: list[dict[str, object]] = [] + + async def log(**kwargs: object) -> None: + logged.append(kwargs) + + handler = TranscribePassthroughLoggingHandler(job_pricer=job_pricer) + logging_obj = _make_logging_obj() + task = handler.schedule_priced_job_logging( + httpx_response=_make_response("StartTranscriptionJob"), + logging_obj=logging_obj, + url_route="https://transcribe.us-west-2.amazonaws.com/", + result='{"TranscriptionJob": {}}', + start_time=datetime.now(), + end_time=datetime.now(), + cache_hit=False, + request_body={"TranscriptionJobName": "litellm-job-1"}, + log=log, + standard_pass_through_logging_payload={"cost_per_request": None}, + ) + await task + + assert priced == [("litellm-job-1", "us-west-2", transcribe_cost_per_second())] + assert len(logged) == 1 + assert logged[0]["response_cost"] == 0.0018 + assert logged[0]["model"] == "transcribe/StartTranscriptionJob" + assert logged[0]["standard_pass_through_logging_payload"] == {"cost_per_request": None} + assert logging_obj.model_call_details["response_cost"] == 0.0018 + + @pytest.mark.asyncio + async def test_job_is_not_logged_for_free_when_the_rate_leaves_the_cost_map(self, monkeypatch: pytest.MonkeyPatch): + async def job_pricer(job_name: str, aws_region_name: str, cost_per_second: float) -> float: + raise AssertionError("pricer must not run without a rate") + + logged: list[dict[str, object]] = [] + + async def log(**kwargs: object) -> None: + logged.append(kwargs) + + monkeypatch.delitem(litellm.model_cost, "transcribe/StartTranscriptionJob") + await TranscribePassthroughLoggingHandler(job_pricer=job_pricer).schedule_priced_job_logging( + httpx_response=_make_response("StartTranscriptionJob"), + logging_obj=_make_logging_obj(), + url_route="https://transcribe.us-west-2.amazonaws.com/", + result='{"TranscriptionJob": {}}', + start_time=datetime.now(), + end_time=datetime.now(), + cache_hit=False, + request_body={"TranscriptionJobName": "litellm-job-1"}, + log=log, + ) + + assert logged == [] + + @pytest.mark.asyncio + async def test_pass_through_success_handler_routes_job_starts_to_the_pricer(self): + scheduled: list[str] = [] + + async def job_pricer(job_name: str, aws_region_name: str, cost_per_second: float) -> float: + scheduled.append(job_name) + return 0.0 + + logging = PassThroughEndpointLogging(TranscribePassthroughLoggingHandler(job_pricer=job_pricer)) + immediate: list[dict[str, object]] = [] + + async def handle_logging(**kwargs: object) -> None: + immediate.append(kwargs) + + logging._handle_logging = handle_logging # rebind-ok: the shared dispatch is the observable under test + + await logging.pass_through_async_success_handler( + httpx_response=_make_response("StartTranscriptionJob"), + response_body={"TranscriptionJob": {}}, + logging_obj=_make_logging_obj(), + url_route="https://transcribe.us-west-2.amazonaws.com/", + result='{"TranscriptionJob": {}}', + start_time=datetime.now(), + end_time=datetime.now(), + cache_hit=False, + request_body={"TranscriptionJobName": "litellm-job-1"}, + passthrough_logging_payload={"url": "https://transcribe.us-west-2.amazonaws.com/"}, + custom_llm_provider="transcribe", + ) + await asyncio.gather(*logging.transcribe_passthrough_logging_handler._pricing_tasks) + + assert scheduled == ["litellm-job-1"] + assert [entry["response_cost"] for entry in immediate] == [0.0] + class TestIsTranscribeRoute: def test_matches_by_provider_tag(self): diff --git a/tests/test_litellm/proxy/pass_through_endpoints/test_llm_pass_through_endpoints.py b/tests/test_litellm/proxy/pass_through_endpoints/test_llm_pass_through_endpoints.py index e447868f454..85c21e7ee60 100644 --- a/tests/test_litellm/proxy/pass_through_endpoints/test_llm_pass_through_endpoints.py +++ b/tests/test_litellm/proxy/pass_through_endpoints/test_llm_pass_through_endpoints.py @@ -5301,7 +5301,9 @@ class TestTranscribeProxyRoute: ) def test_signs_and_forwards_start_transcription_job(self, transcribe_client: TestClient) -> None: - upstream_body = {"TranscriptionJob": {"TranscriptionJobName": "litellm-job-1", "TranscriptionJobStatus": "IN_PROGRESS"}} + upstream_body = { + "TranscriptionJob": {"TranscriptionJobName": "litellm-job-1", "TranscriptionJobStatus": "IN_PROGRESS"} + } with respx.mock(assert_all_called=True) as upstream: route = upstream.post(TRANSCRIBE_UPSTREAM).mock(return_value=httpx.Response(200, json=upstream_body)) response = transcribe_client.post( @@ -5311,9 +5313,11 @@ class TestTranscribeProxyRoute: ) assert (response.status_code, response.json()) == (200, upstream_body) - sent = route.calls.last.request + targets = [call.request.headers["x-amz-target"] for call in route.calls] + assert targets[0] == "Transcribe.StartTranscriptionJob" + assert set(targets[1:]) <= {"Transcribe.GetTranscriptionJob"} + sent = route.calls[0].request assert json.loads(sent.content) == dict(self.START_JOB_BODY) - assert sent.headers["x-amz-target"] == "Transcribe.StartTranscriptionJob" assert sent.headers["content-type"] == "application/x-amz-json-1.1" assert sent.headers["authorization"].startswith("AWS4-HMAC-SHA256 Credential=test-access-key/") assert "/us-west-2/transcribe/aws4_request" in sent.headers["authorization"] @@ -5334,7 +5338,10 @@ class TestTranscribeProxyRoute: }, ) - assert (response.status_code, response.json()) == (200, {"TranscriptionJob": {"TranscriptionJobStatus": "COMPLETED"}}) + assert (response.status_code, response.json()) == ( + 200, + {"TranscriptionJob": {"TranscriptionJobStatus": "COMPLETED"}}, + ) sent = route.calls.last.request assert sent.headers["x-amz-target"] == "Transcribe.GetTranscriptionJob" assert "Credential=test-access-key/" in sent.headers["authorization"] @@ -5344,15 +5351,25 @@ class TestTranscribeProxyRoute: aws_error = {"__type": "BadRequestException", "Message": "The requested job couldn't be found."} with respx.mock(assert_all_called=True) as upstream: upstream.post(TRANSCRIBE_UPSTREAM).mock(return_value=httpx.Response(400, json=aws_error)) - response = transcribe_client.post("/transcribe/GetTranscriptionJob", json={"TranscriptionJobName": "missing"}) + response = transcribe_client.post( + "/transcribe/GetTranscriptionJob", json={"TranscriptionJobName": "missing"} + ) assert (response.status_code, response.json()) == (400, aws_error) @pytest.mark.parametrize( "operation", - ["Start-Transcription-Job", "Transcribe.StartTranscriptionJob", "a" * 200, "starttranscriptionjob", "DetectEntitiesV2"], + [ + "Start-Transcription-Job", + "Transcribe.StartTranscriptionJob", + "a" * 200, + "starttranscriptionjob", + "DetectEntitiesV2", + ], ) - def test_rejects_unsupported_operations_without_calling_aws(self, transcribe_client: TestClient, operation: str) -> None: + def test_rejects_unsupported_operations_without_calling_aws( + self, transcribe_client: TestClient, operation: str + ) -> None: with respx.mock(assert_all_called=False) as upstream: route = upstream.post(TRANSCRIBE_UPSTREAM) response = transcribe_client.post(f"/transcribe/{operation}", json={}) @@ -5388,6 +5405,40 @@ class TestTranscribeProxyRoute: assert "AWS region" in response.json()["detail"] assert not route.called + @pytest.mark.parametrize( + ("operation", "body", "detail_fragment"), + [ + ("StartMedicalTranscriptionJob", {"MedicalTranscriptionJobName": "j"}, "StartMedicalTranscriptionJob"), + ("StartCallAnalyticsJob", {"CallAnalyticsJobName": "j"}, "StartCallAnalyticsJob"), + ("StartMedicalScribeJob", {"MedicalScribeJobName": "j"}, "StartMedicalScribeJob"), + ("StartTranscriptionJob", {"ContentRedaction": {"RedactionType": "PII"}}, "ContentRedaction"), + ("StartTranscriptionJob", {"ToxicityDetection": [{"ToxicityCategories": ["ALL"]}]}, "ToxicityDetection"), + ("StartTranscriptionJob", {"ModelSettings": {"LanguageModelName": "clm"}}, "LanguageModelName"), + ], + ) + def test_rejects_unpriced_billable_jobs_without_calling_aws( + self, transcribe_client: TestClient, operation: str, body: dict[str, object], detail_fragment: str + ) -> None: + with respx.mock(assert_all_called=False) as upstream: + route = upstream.post(TRANSCRIBE_UPSTREAM) + response = transcribe_client.post(f"/transcribe/{operation}", json={**dict(self.START_JOB_BODY), **body}) + + assert response.status_code == 400 + assert detail_fragment in response.json()["detail"] + assert not route.called + + def test_rejects_start_transcription_job_when_the_cost_map_has_no_rate( + self, transcribe_client: TestClient, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.delitem(litellm.model_cost, "transcribe/StartTranscriptionJob") + with respx.mock(assert_all_called=False) as upstream: + route = upstream.post(TRANSCRIBE_UPSTREAM) + response = transcribe_client.post("/transcribe/StartTranscriptionJob", json=dict(self.START_JOB_BODY)) + + assert response.status_code == 400 + assert "model cost map" in response.json()["detail"] + assert not route.called + @pytest.mark.parametrize("target_header", ["", "Transcribe", "ComprehendMedical_20181030.DetectPHI", "Transcribe."]) def test_sdk_route_rejects_bad_x_amz_target(self, transcribe_client: TestClient, target_header: str) -> None: with respx.mock(assert_all_called=False) as upstream: