From 784fe5bfd873f6d513210ed2250d0fc2d8557901 Mon Sep 17 00:00:00 2001 From: yassin Date: Thu, 17 Sep 2026 19:11:45 +0000 Subject: [PATCH] fix(proxy): price Amazon Transcribe jobs at completion so budgets apply StartTranscriptionJob was logged with response_cost 0.0, so key, team and proxy budgets never stopped repeated jobs on the proxy's AWS credentials. The success handler now polls GetTranscriptionJob to completion, reads the audio duration from the transcript artifact and charges whole seconds at the cost map rate, charging the longest media AWS accepts when the duration cannot be read. The route refuses job classes and surcharge features the cost map does not price Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/constants.py | 4 + ...odel_prices_and_context_window_backup.json | 10 + .../llm_passthrough_endpoints.py | 5 + .../transcribe_passthrough_logging_handler.py | 357 +++++++++++++++++- .../pass_through_endpoints/success_handler.py | 22 +- model_prices_and_context_window.json | 10 + ..._transcribe_passthrough_logging_handler.py | 284 +++++++++++++- .../test_llm_pass_through_endpoints.py | 65 +++- 8 files changed, 733 insertions(+), 24 deletions(-) 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: