mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-21 00:21:49 +00:00
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>
This commit is contained in:
yassin
parent
65d0f3a03d
commit
784fe5bfd8
8 changed files with 733 additions and 24 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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):
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue