mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-03 02:22:24 +00:00
Add gpt-realtime-whisper Realtime transcription support (OpenAI + Azure)
Adds first-class support for the gpt-realtime-whisper streaming speech-to-text
model, which uses the Realtime transcription session API rather than the
file-based /audio/transcriptions path.
Model registration: registers gpt-realtime-whisper and azure/gpt-realtime-whisper
with audio-duration pricing (input_cost_per_second = 0.017/60, matching the
published $0.017/minute input audio rate).
REST endpoint: implements POST /v1/realtime/transcription_sessions (plus /realtime
and /openai/v1 aliases) to mint an ephemeral transcription session for the
WebRTC flow. Adds request/response types, OpenAI and Azure URL builders, a shared
base handler (refactored from the client_secrets handler), the
acreate_realtime_transcription_session SDK function, and route registration. The
proxy encrypts the ephemeral key returned under client_secret.value and records
the session type in the token so the follow-up /realtime/calls replays
type=transcription rather than type=realtime.
WebSocket: forwards intent=transcription through to the Azure handler (OpenAI
already received it) with URL-encoding, so gpt-realtime-whisper opens a
transcription session. Transcription-only sessions no longer trigger an
erroneous response.create.
Cost tracking: transcription sessions emit no response.done events; their usage
arrives on conversation.item.input_audio_transcription.completed as
{type: duration, seconds}. That usage is captured out-of-band (usage only, no
transcript duplication) and billed by input_cost_per_second, with a token-billed
fallback for token-priced transcription models.
Adds tests for pricing math, URL builders, request/response types, the proxy
route and SDK function, WebSocket intent forwarding, transcription-session
streaming behavior, and the /realtime/calls session-type replay.
This commit is contained in:
Emerson Gomes
parent
1c741b91c0
commit
63d754d27b
21 changed files with 1324 additions and 11 deletions
|
|
@ -1350,6 +1350,7 @@ from .search.main import *
|
|||
from .realtime_api.main import (
|
||||
_arealtime,
|
||||
acreate_realtime_client_secret,
|
||||
acreate_realtime_transcription_session,
|
||||
arealtime_calls,
|
||||
)
|
||||
from .responses.main import _aresponses_websocket
|
||||
|
|
|
|||
|
|
@ -2534,4 +2534,101 @@ def handle_realtime_stream_cost_calculation(
|
|||
break # exit if we find a valid model
|
||||
total_cost = input_cost_per_token + output_cost_per_token
|
||||
|
||||
total_cost += handle_realtime_transcription_cost_calculation(
|
||||
results=results,
|
||||
custom_llm_provider=custom_llm_provider,
|
||||
litellm_model_name=litellm_model_name,
|
||||
)
|
||||
|
||||
return total_cost
|
||||
|
||||
|
||||
_TRANSCRIPTION_COMPLETED_EVENT_TYPE = (
|
||||
"conversation.item.input_audio_transcription.completed"
|
||||
)
|
||||
|
||||
|
||||
def handle_realtime_transcription_cost_calculation(
|
||||
results: OpenAIRealtimeStreamList,
|
||||
custom_llm_provider: str,
|
||||
litellm_model_name: str,
|
||||
) -> float:
|
||||
"""
|
||||
Cost for realtime transcription sessions (e.g. gpt-realtime-whisper).
|
||||
|
||||
Transcription sessions emit no `response.done` events; instead each
|
||||
`conversation.item.input_audio_transcription.completed` event carries a
|
||||
`usage` object billed by the ASR model. The usage is one of:
|
||||
- {"type": "duration", "seconds": <float>} → priced via input_cost_per_second
|
||||
- {"type": "tokens", "input_tokens": ...} → priced via input/audio token cost
|
||||
"""
|
||||
completed_events = [
|
||||
cast(dict, result)
|
||||
for result in results
|
||||
if result.get("type") == _TRANSCRIPTION_COMPLETED_EVENT_TYPE
|
||||
]
|
||||
if not completed_events:
|
||||
return 0.0
|
||||
|
||||
model_name = (
|
||||
_get_transcription_model_name_from_results(results) or litellm_model_name
|
||||
)
|
||||
try:
|
||||
model_info = litellm.get_model_info(
|
||||
model=model_name, custom_llm_provider=custom_llm_provider
|
||||
)
|
||||
except Exception:
|
||||
model_info = {}
|
||||
|
||||
total_cost = 0.0
|
||||
for event in completed_events:
|
||||
usage = event.get("usage") or {}
|
||||
total_cost += _transcription_usage_cost(usage, model_info)
|
||||
return total_cost
|
||||
|
||||
|
||||
def _get_transcription_model_name_from_results(
|
||||
results: OpenAIRealtimeStreamList,
|
||||
) -> Optional[str]:
|
||||
"""Resolve the ASR model from a transcription_session.* / session.* event."""
|
||||
for result in results:
|
||||
if result.get("type") in (
|
||||
"transcription_session.created",
|
||||
"transcription_session.updated",
|
||||
"session.created",
|
||||
"session.updated",
|
||||
):
|
||||
session = cast(dict, result).get("session", {}) or {}
|
||||
transcription = (
|
||||
(session.get("audio", {}) or {}).get("input", {}) or {}
|
||||
).get("transcription", {}) or session.get("input_audio_transcription", {})
|
||||
model = (transcription or {}).get("model")
|
||||
if model:
|
||||
return model
|
||||
return None
|
||||
|
||||
|
||||
def _transcription_usage_cost(usage: dict, model_info: dict) -> float:
|
||||
usage_type = usage.get("type")
|
||||
if usage_type == "duration":
|
||||
seconds = usage.get("seconds") or 0.0
|
||||
per_second = model_info.get("input_cost_per_second") or 0.0
|
||||
return float(seconds) * float(per_second)
|
||||
if usage_type == "tokens":
|
||||
input_token_details = usage.get("input_token_details") or {}
|
||||
audio_tokens = input_token_details.get("audio_tokens") or 0
|
||||
text_tokens = input_token_details.get("text_tokens") or 0
|
||||
output_tokens = usage.get("output_tokens") or 0
|
||||
audio_cost = float(audio_tokens) * float(
|
||||
model_info.get("input_cost_per_audio_token")
|
||||
or model_info.get("input_cost_per_token")
|
||||
or 0.0
|
||||
)
|
||||
text_cost = float(text_tokens) * float(
|
||||
model_info.get("input_cost_per_token") or 0.0
|
||||
)
|
||||
output_cost = float(output_tokens) * float(
|
||||
model_info.get("output_cost_per_token") or 0.0
|
||||
)
|
||||
return audio_cost + text_cost + output_cost
|
||||
return 0.0
|
||||
|
|
|
|||
|
|
@ -100,6 +100,10 @@ class RealTimeStreaming:
|
|||
self._flushing_pending_messages_until_setup: bool = False
|
||||
self._pending_messages_until_setup: List[str] = []
|
||||
self._pending_messages_byte_total: int = 0
|
||||
# Whether this is a transcription-only session (session.type == "transcription",
|
||||
# e.g. gpt-realtime-whisper). Such sessions must not be sent response.create and
|
||||
# their input_audio_transcription.completed usage drives duration-based cost.
|
||||
self._is_transcription_session: bool = False
|
||||
|
||||
# Per-connection caps for pre-setup audio frames (message count + total bytes).
|
||||
_MAX_BUFFERED_MESSAGES: int = 200
|
||||
|
|
@ -209,6 +213,8 @@ class RealTimeStreaming:
|
|||
self.session_tools = tools
|
||||
# GA: session.type is required; log it for traceability but no action needed
|
||||
verbose_logger.debug(f"Realtime session.type: {session.get('type')}")
|
||||
if session.get("type") == "transcription":
|
||||
self._is_transcription_session = True
|
||||
except (json.JSONDecodeError, AttributeError, TypeError):
|
||||
pass
|
||||
|
||||
|
|
@ -225,6 +231,55 @@ class RealTimeStreaming:
|
|||
except (AttributeError, TypeError):
|
||||
pass
|
||||
|
||||
def _detect_transcription_session_from_backend(
|
||||
self, event_obj: Union[dict, OpenAIRealtimeEvents]
|
||||
) -> None:
|
||||
"""Flag transcription-only sessions from backend session events."""
|
||||
try:
|
||||
event_type = event_obj.get("type", "")
|
||||
if event_type in (
|
||||
"transcription_session.created",
|
||||
"transcription_session.updated",
|
||||
):
|
||||
self._is_transcription_session = True
|
||||
elif event_type in ("session.created", "session.updated"):
|
||||
session = cast(dict, event_obj).get("session", {}) or {}
|
||||
if session.get("type") == "transcription":
|
||||
self._is_transcription_session = True
|
||||
except (AttributeError, TypeError):
|
||||
pass
|
||||
|
||||
def _capture_transcription_usage(
|
||||
self, event_obj: Union[dict, OpenAIRealtimeEvents]
|
||||
) -> None:
|
||||
"""
|
||||
Append a usage-only transcription completed event to the logged results so
|
||||
the cost calculator can bill it by audio duration. The default logged event
|
||||
types exclude this event, so it is captured here directly for transcription
|
||||
sessions rather than widening logging for every realtime session. Only the
|
||||
type and usage are kept — the transcript is already captured separately in
|
||||
input_messages, so it is not duplicated into the response log here.
|
||||
"""
|
||||
try:
|
||||
usage = event_obj.get("usage")
|
||||
if usage is None:
|
||||
return
|
||||
# If this event type is already captured by store_message (e.g. the user
|
||||
# logs all realtime events), don't append a second copy.
|
||||
if self._should_store_message(event_obj):
|
||||
return
|
||||
self.messages.append(
|
||||
cast(
|
||||
OpenAIRealtimeEvents,
|
||||
{
|
||||
"type": "conversation.item.input_audio_transcription.completed",
|
||||
"usage": usage,
|
||||
},
|
||||
)
|
||||
)
|
||||
except (AttributeError, TypeError):
|
||||
pass
|
||||
|
||||
def _collect_tool_calls_from_response_done(
|
||||
self, event_obj: Union[dict, OpenAIRealtimeEvents]
|
||||
) -> None:
|
||||
|
|
@ -713,6 +768,8 @@ class RealTimeStreaming:
|
|||
try:
|
||||
event_obj = json.loads(raw_response)
|
||||
|
||||
self._detect_transcription_session_from_backend(event_obj)
|
||||
|
||||
# For audio/VAD guardrail path: once the session is ready, tell the backend
|
||||
# not to auto-respond after VAD detects end-of-speech. We send the
|
||||
# session.created to the client FIRST so the client is always in sync, then
|
||||
|
|
@ -731,12 +788,20 @@ class RealTimeStreaming:
|
|||
event_obj.get("type")
|
||||
== "conversation.item.input_audio_transcription.completed"
|
||||
):
|
||||
transcript = event_obj.get("transcript", "")
|
||||
self._collect_user_input_from_backend_event(event_obj)
|
||||
## LOGGING — must happen before continue below
|
||||
self.store_message(raw_response)
|
||||
# Forward transcript to client so user sees what they said
|
||||
await self.websocket.send_text(raw_response)
|
||||
|
||||
# Transcription-only sessions (e.g. gpt-realtime-whisper) have no
|
||||
# assistant turn: capture audio-duration usage for cost and never
|
||||
# trigger response.create.
|
||||
if self._is_transcription_session:
|
||||
self._capture_transcription_usage(event_obj)
|
||||
return True
|
||||
|
||||
transcript = event_obj.get("transcript", "")
|
||||
blocked = await self.run_realtime_guardrails(
|
||||
transcript,
|
||||
item_id=event_obj.get("item_id"),
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ This requires websockets, and is currently only supported on LiteLLM Proxy.
|
|||
"""
|
||||
|
||||
from typing import Any, Optional, cast
|
||||
from urllib.parse import quote
|
||||
|
||||
from litellm._logging import _redact_string, verbose_proxy_logger
|
||||
from litellm.constants import REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES
|
||||
|
|
@ -35,6 +36,7 @@ class AzureOpenAIRealtime(AzureChatCompletion):
|
|||
model: str,
|
||||
api_version: Optional[str],
|
||||
realtime_protocol: Optional[str] = None,
|
||||
query_params: Optional[dict] = None,
|
||||
) -> str:
|
||||
"""
|
||||
Construct Azure realtime WebSocket URL.
|
||||
|
|
@ -46,6 +48,7 @@ class AzureOpenAIRealtime(AzureChatCompletion):
|
|||
realtime_protocol: Protocol version to use:
|
||||
- "GA" or "v1": Uses /openai/v1/realtime (GA path)
|
||||
- "beta" or None: Uses /openai/realtime (beta path, default)
|
||||
query_params: Extra query params to forward (e.g. intent=transcription).
|
||||
|
||||
Returns:
|
||||
WebSocket URL string
|
||||
|
|
@ -63,11 +66,16 @@ class AzureOpenAIRealtime(AzureChatCompletion):
|
|||
)
|
||||
if _is_ga:
|
||||
path = "/openai/v1/realtime"
|
||||
return f"{api_base}{path}?model={model}"
|
||||
url = f"{api_base}{path}?model={model}"
|
||||
else:
|
||||
# Default to beta path for backwards compatibility
|
||||
path = "/openai/realtime"
|
||||
return f"{api_base}{path}?api-version={api_version}&deployment={model}"
|
||||
url = f"{api_base}{path}?api-version={api_version}&deployment={model}"
|
||||
|
||||
intent = (query_params or {}).get("intent")
|
||||
if intent:
|
||||
url = f"{url}&intent={quote(str(intent), safe='')}"
|
||||
return url
|
||||
|
||||
async def async_realtime(
|
||||
self,
|
||||
|
|
@ -81,6 +89,7 @@ class AzureOpenAIRealtime(AzureChatCompletion):
|
|||
client: Optional[Any] = None,
|
||||
timeout: Optional[float] = None,
|
||||
realtime_protocol: Optional[str] = None,
|
||||
query_params: Optional[dict] = None,
|
||||
user_api_key_dict: Optional[Any] = None,
|
||||
litellm_metadata: Optional[dict] = None,
|
||||
):
|
||||
|
|
@ -96,7 +105,11 @@ class AzureOpenAIRealtime(AzureChatCompletion):
|
|||
raise ValueError("api_version is required for Azure OpenAI calls")
|
||||
|
||||
url = self._construct_url(
|
||||
api_base, model, api_version, realtime_protocol=realtime_protocol
|
||||
api_base,
|
||||
model,
|
||||
api_version,
|
||||
realtime_protocol=realtime_protocol,
|
||||
query_params=query_params,
|
||||
)
|
||||
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -40,6 +40,13 @@ class AzureRealtimeHTTPConfig(BaseRealtimeHTTPConfig):
|
|||
version = api_version or get_secret_str("AZURE_API_VERSION") or "2024-12-17"
|
||||
return f"{base}/openai/realtime/calls?api-version={version}"
|
||||
|
||||
def get_transcription_session_url(
|
||||
self, api_base: Optional[str], model: str, api_version: Optional[str] = None
|
||||
) -> str:
|
||||
base = self.get_api_base(api_base).rstrip("/")
|
||||
version = api_version or get_secret_str("AZURE_API_VERSION") or "2024-12-17"
|
||||
return f"{base}/openai/realtime/transcription_sessions?api-version={version}"
|
||||
|
||||
def get_realtime_calls_headers(self, ephemeral_key: str) -> dict:
|
||||
return {
|
||||
"api-key": ephemeral_key,
|
||||
|
|
|
|||
|
|
@ -59,6 +59,15 @@ class BaseRealtimeHTTPConfig(ABC):
|
|||
) -> str:
|
||||
"""Return the full URL for POST /realtime/client_secrets."""
|
||||
|
||||
def get_transcription_session_url(
|
||||
self, api_base: Optional[str], model: str, api_version: Optional[str] = None
|
||||
) -> str:
|
||||
"""Return the full URL for POST /realtime/transcription_sessions."""
|
||||
base = (api_base or "").rstrip("/")
|
||||
if base.endswith("/v1"):
|
||||
base = base[:-3]
|
||||
return f"{base}/v1/realtime/transcription_sessions"
|
||||
|
||||
@abstractmethod
|
||||
def validate_environment(
|
||||
self,
|
||||
|
|
|
|||
|
|
@ -5382,6 +5382,69 @@ class BaseLLMHTTPHandler:
|
|||
"""
|
||||
Forward POST /v1/realtime/client_secrets to upstream provider.
|
||||
|
||||
Uses provider_config (BaseRealtimeHTTPConfig) for URL construction and
|
||||
header auth when available; falls back to the legacy OpenAI-style defaults.
|
||||
"""
|
||||
return await self._async_realtime_session_post(
|
||||
endpoint="client_secrets",
|
||||
api_base=api_base,
|
||||
api_key=api_key,
|
||||
request_data=request_data,
|
||||
logging_obj=logging_obj,
|
||||
timeout=timeout,
|
||||
provider_config=provider_config,
|
||||
model=model,
|
||||
extra_headers=extra_headers,
|
||||
client=client,
|
||||
api_version=api_version,
|
||||
)
|
||||
|
||||
async def async_realtime_transcription_session_handler(
|
||||
self,
|
||||
api_base: str,
|
||||
api_key: str,
|
||||
request_data: Dict[str, Any],
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
timeout: Union[float, httpx.Timeout],
|
||||
provider_config: Optional[Any] = None,
|
||||
model: Optional[str] = None,
|
||||
extra_headers: Optional[Dict[str, Any]] = None,
|
||||
client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
|
||||
api_version: Optional[str] = None,
|
||||
) -> httpx.Response:
|
||||
"""Forward POST /v1/realtime/transcription_sessions to upstream provider."""
|
||||
return await self._async_realtime_session_post(
|
||||
endpoint="transcription_sessions",
|
||||
api_base=api_base,
|
||||
api_key=api_key,
|
||||
request_data=request_data,
|
||||
logging_obj=logging_obj,
|
||||
timeout=timeout,
|
||||
provider_config=provider_config,
|
||||
model=model,
|
||||
extra_headers=extra_headers,
|
||||
client=client,
|
||||
api_version=api_version,
|
||||
)
|
||||
|
||||
async def _async_realtime_session_post(
|
||||
self,
|
||||
endpoint: Literal["client_secrets", "transcription_sessions"],
|
||||
api_base: str,
|
||||
api_key: str,
|
||||
request_data: Dict[str, Any],
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
timeout: Union[float, httpx.Timeout],
|
||||
provider_config: Optional[Any] = None,
|
||||
model: Optional[str] = None,
|
||||
extra_headers: Optional[Dict[str, Any]] = None,
|
||||
client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
|
||||
api_version: Optional[str] = None,
|
||||
) -> httpx.Response:
|
||||
"""
|
||||
Shared POST flow for the realtime HTTP session endpoints
|
||||
(client_secrets and transcription_sessions).
|
||||
|
||||
Uses provider_config (BaseRealtimeHTTPConfig) for URL construction and
|
||||
header auth when available; falls back to the legacy OpenAI-style defaults.
|
||||
"""
|
||||
|
|
@ -5393,14 +5456,19 @@ class BaseLLMHTTPHandler:
|
|||
async_httpx_client = client
|
||||
|
||||
if provider_config is not None:
|
||||
url = provider_config.get_complete_url(
|
||||
api_base=api_base, model=model or "", api_version=api_version
|
||||
)
|
||||
if endpoint == "transcription_sessions":
|
||||
url = provider_config.get_transcription_session_url(
|
||||
api_base=api_base, model=model or "", api_version=api_version
|
||||
)
|
||||
else:
|
||||
url = provider_config.get_complete_url(
|
||||
api_base=api_base, model=model or "", api_version=api_version
|
||||
)
|
||||
headers: Dict[str, Any] = provider_config.validate_environment(
|
||||
headers={}, model=model or "", api_key=api_key
|
||||
)
|
||||
else:
|
||||
url = f"{api_base.rstrip('/')}/v1/realtime/client_secrets"
|
||||
url = f"{api_base.rstrip('/')}/v1/realtime/{endpoint}"
|
||||
headers = {
|
||||
"Authorization": f"Bearer {api_key}",
|
||||
"Content-Type": "application/json",
|
||||
|
|
|
|||
|
|
@ -41,6 +41,14 @@ class OpenAIRealtimeHTTPConfig(BaseRealtimeHTTPConfig):
|
|||
base = base[:-3]
|
||||
return f"{base}/v1/realtime/calls"
|
||||
|
||||
def get_transcription_session_url(
|
||||
self, api_base: Optional[str], model: str, api_version: Optional[str] = None
|
||||
) -> str:
|
||||
base = self.get_api_base(api_base).rstrip("/")
|
||||
if base.endswith("/v1"):
|
||||
base = base[:-3]
|
||||
return f"{base}/v1/realtime/transcription_sessions"
|
||||
|
||||
def validate_environment(
|
||||
self,
|
||||
headers: dict,
|
||||
|
|
|
|||
|
|
@ -4204,6 +4204,23 @@
|
|||
"/v1/audio/transcriptions"
|
||||
]
|
||||
},
|
||||
"azure/gpt-realtime-whisper": {
|
||||
"input_cost_per_second": 0.0002833333333333333,
|
||||
"litellm_provider": "azure",
|
||||
"mode": "audio_transcription",
|
||||
"source": "https://learn.microsoft.com/en-us/azure/ai-foundry/openai/concepts/gpt-realtime-whisper",
|
||||
"supported_endpoints": [
|
||||
"/v1/realtime",
|
||||
"/v1/realtime/transcription_sessions"
|
||||
],
|
||||
"supported_modalities": [
|
||||
"audio"
|
||||
],
|
||||
"supported_output_modalities": [
|
||||
"text"
|
||||
],
|
||||
"supports_audio_input": true
|
||||
},
|
||||
"azure/gpt-5.1-2025-11-13": {
|
||||
"cache_read_input_token_cost": 1.25e-07,
|
||||
"cache_read_input_token_cost_priority": 2.5e-07,
|
||||
|
|
@ -40507,6 +40524,23 @@
|
|||
"supports_system_messages": true,
|
||||
"supports_tool_choice": true
|
||||
},
|
||||
"gpt-realtime-whisper": {
|
||||
"input_cost_per_second": 0.0002833333333333333,
|
||||
"litellm_provider": "openai",
|
||||
"mode": "audio_transcription",
|
||||
"source": "https://platform.openai.com/docs/models/gpt-realtime-whisper",
|
||||
"supported_endpoints": [
|
||||
"/v1/realtime",
|
||||
"/v1/realtime/transcription_sessions"
|
||||
],
|
||||
"supported_modalities": [
|
||||
"audio"
|
||||
],
|
||||
"supported_output_modalities": [
|
||||
"text"
|
||||
],
|
||||
"supports_audio_input": true
|
||||
},
|
||||
"sora-2": {
|
||||
"litellm_provider": "openai",
|
||||
"mode": "video_generation",
|
||||
|
|
|
|||
|
|
@ -19,6 +19,8 @@ from litellm.proxy.common_utils.http_parsing_utils import _read_request_body
|
|||
from litellm.types.realtime import (
|
||||
RealtimeClientSecretRequest,
|
||||
RealtimeClientSecretResponse,
|
||||
RealtimeTranscriptionSessionRequest,
|
||||
RealtimeTranscriptionSessionResponse,
|
||||
)
|
||||
|
||||
router = APIRouter()
|
||||
|
|
@ -32,6 +34,7 @@ def _encode_realtime_token_payload(
|
|||
user_id: Optional[str],
|
||||
team_id: Optional[str],
|
||||
expires_at: Optional[int],
|
||||
session_type: str = "realtime",
|
||||
) -> str:
|
||||
"""
|
||||
Encode metadata with the upstream ephemeral key so /realtime/calls can
|
||||
|
|
@ -44,6 +47,7 @@ def _encode_realtime_token_payload(
|
|||
"user_id": user_id or "",
|
||||
"team_id": team_id or "",
|
||||
"expires_at": expires_at,
|
||||
"session_type": session_type,
|
||||
}
|
||||
return json.dumps(payload, separators=(",", ":"))
|
||||
|
||||
|
|
@ -199,6 +203,9 @@ async def create_realtime_client_secret(
|
|||
user_id=getattr(user_api_key_dict, "user_id", None),
|
||||
team_id=getattr(user_api_key_dict, "team_id", None),
|
||||
expires_at=expires_at if isinstance(expires_at, int) else None,
|
||||
session_type=(
|
||||
req.session.type if req.session and req.session.type else "realtime"
|
||||
),
|
||||
)
|
||||
encrypted_token: str = encrypt_value_helper(token_payload)
|
||||
upstream_json["value"] = encrypted_token
|
||||
|
|
@ -283,12 +290,14 @@ async def proxy_realtime_calls(
|
|||
)
|
||||
user_id = decoded_payload.get("user_id") or None
|
||||
team_id = decoded_payload.get("team_id") or None
|
||||
session_type = decoded_payload.get("session_type") or "realtime"
|
||||
else:
|
||||
# Backward compatibility: older tokens contained only encrypted upstream key.
|
||||
openai_ephemeral_key = decrypted_token_value
|
||||
model = request.query_params.get("model", "gpt-4o-realtime-preview")
|
||||
user_id = None
|
||||
team_id = None
|
||||
session_type = "realtime"
|
||||
|
||||
# Build a minimal UserAPIKeyAuth with user/team IDs from the token
|
||||
# so spend tracking and budget enforcement work correctly.
|
||||
|
|
@ -301,7 +310,7 @@ async def proxy_realtime_calls(
|
|||
try:
|
||||
# Build session config for the multipart form data
|
||||
session_config = {
|
||||
"type": "realtime",
|
||||
"type": session_type,
|
||||
"model": model,
|
||||
}
|
||||
|
||||
|
|
@ -366,3 +375,136 @@ async def proxy_realtime_calls(
|
|||
status_code=upstream_resp.status_code,
|
||||
media_type=upstream_resp.headers.get("content-type", "application/sdp"),
|
||||
)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/v1/realtime/transcription_sessions",
|
||||
dependencies=[Depends(user_api_key_auth)],
|
||||
tags=["realtime"],
|
||||
)
|
||||
@router.post(
|
||||
"/realtime/transcription_sessions",
|
||||
dependencies=[Depends(user_api_key_auth)],
|
||||
tags=["realtime"],
|
||||
)
|
||||
@router.post(
|
||||
"/openai/v1/realtime/transcription_sessions",
|
||||
dependencies=[Depends(user_api_key_auth)],
|
||||
tags=["realtime"],
|
||||
)
|
||||
async def create_realtime_transcription_session(
|
||||
request: Request,
|
||||
fastapi_response: Response,
|
||||
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
|
||||
) -> RealtimeTranscriptionSessionResponse:
|
||||
"""
|
||||
Create an ephemeral Realtime transcription session
|
||||
(POST /v1/realtime/transcription_sessions) for the WebRTC/WebSocket flow.
|
||||
|
||||
Mirrors the client_secrets route but targets the transcription_sessions
|
||||
endpoint and encrypts the ephemeral key returned under `client_secret.value`.
|
||||
"""
|
||||
from litellm.proxy.proxy_server import (
|
||||
add_litellm_data_to_request,
|
||||
general_settings,
|
||||
llm_router,
|
||||
proxy_config,
|
||||
proxy_logging_obj,
|
||||
route_request,
|
||||
user_model,
|
||||
version,
|
||||
)
|
||||
|
||||
data: dict = {}
|
||||
try:
|
||||
body = await _read_request_body(request=request)
|
||||
req = RealtimeTranscriptionSessionRequest(**body)
|
||||
|
||||
model: str = req.resolved_model() or "gpt-realtime-whisper"
|
||||
|
||||
transcription_session = {k: v for k, v in body.items() if k != "model"}
|
||||
data = {"model": model, "transcription_session": transcription_session}
|
||||
|
||||
data = await add_litellm_data_to_request(
|
||||
data=data,
|
||||
request=request,
|
||||
general_settings=general_settings,
|
||||
user_api_key_dict=user_api_key_dict,
|
||||
version=version,
|
||||
proxy_config=proxy_config,
|
||||
)
|
||||
|
||||
data = await proxy_logging_obj.pre_call_hook(
|
||||
user_api_key_dict=user_api_key_dict,
|
||||
data=data,
|
||||
call_type="acreate_realtime_transcription_session",
|
||||
)
|
||||
|
||||
verbose_proxy_logger.debug(
|
||||
"Realtime: /v1/realtime/transcription_sessions (model=%s)", model
|
||||
)
|
||||
|
||||
llm_call = await route_request(
|
||||
data=data,
|
||||
route_type="acreate_realtime_transcription_session",
|
||||
llm_router=llm_router,
|
||||
user_model=user_model,
|
||||
)
|
||||
upstream_resp: httpx.Response = await llm_call # type: ignore
|
||||
|
||||
except Exception as e:
|
||||
await proxy_logging_obj.post_call_failure_hook(
|
||||
user_api_key_dict=user_api_key_dict,
|
||||
original_exception=e,
|
||||
request_data=data,
|
||||
)
|
||||
verbose_proxy_logger.error(
|
||||
"litellm.proxy.realtime_endpoints.create_realtime_transcription_session(): Exception - %s",
|
||||
str(e),
|
||||
)
|
||||
if isinstance(e, HTTPException):
|
||||
raise ProxyException(
|
||||
message=getattr(e, "message", str(e)),
|
||||
type=getattr(e, "type", "None"),
|
||||
param=getattr(e, "param", "None"),
|
||||
code=getattr(e, "status_code", http_status.HTTP_400_BAD_REQUEST),
|
||||
)
|
||||
raise ProxyException(
|
||||
message=getattr(e, "message", str(e)),
|
||||
type=getattr(e, "type", "None"),
|
||||
param=getattr(e, "param", "None"),
|
||||
code=getattr(e, "status_code", 500),
|
||||
)
|
||||
|
||||
if upstream_resp.status_code != 200:
|
||||
verbose_proxy_logger.error(
|
||||
"Realtime transcription_sessions upstream error %s: %s",
|
||||
upstream_resp.status_code,
|
||||
upstream_resp.text,
|
||||
)
|
||||
return Response( # type: ignore[return-value]
|
||||
content=upstream_resp.content,
|
||||
status_code=upstream_resp.status_code,
|
||||
media_type="application/json",
|
||||
)
|
||||
|
||||
upstream_json: dict = upstream_resp.json()
|
||||
|
||||
# Encrypt the ephemeral key (returned under client_secret.value) with routing
|
||||
# metadata so the follow-up /realtime/calls request can recover the model.
|
||||
client_secret = upstream_json.get("client_secret")
|
||||
if isinstance(client_secret, dict) and "value" in client_secret:
|
||||
raw_value: str = client_secret.get("value", "")
|
||||
expires_at = client_secret.get("expires_at")
|
||||
token_payload = _encode_realtime_token_payload(
|
||||
ephemeral_key=raw_value,
|
||||
model_id=model,
|
||||
user_id=getattr(user_api_key_dict, "user_id", None),
|
||||
team_id=getattr(user_api_key_dict, "team_id", None),
|
||||
expires_at=expires_at if isinstance(expires_at, int) else None,
|
||||
session_type="transcription",
|
||||
)
|
||||
client_secret["value"] = encrypt_value_helper(token_payload)
|
||||
upstream_json["client_secret"] = client_secret
|
||||
|
||||
return RealtimeTranscriptionSessionResponse(**upstream_json)
|
||||
|
|
|
|||
|
|
@ -74,6 +74,7 @@ ROUTE_ENDPOINT_MAPPING = {
|
|||
"avideo_extension": "/videos/extensions",
|
||||
"acreate_realtime_client_secret": "/realtime/client_secrets",
|
||||
"arealtime_calls": "/realtime/calls",
|
||||
"acreate_realtime_transcription_session": "/realtime/transcription_sessions",
|
||||
"acreate_container": "/containers",
|
||||
"alist_containers": "/containers",
|
||||
"aretrieve_container": "/containers/{container_id}",
|
||||
|
|
@ -261,6 +262,7 @@ async def route_request( # noqa: PLR0915 - Complex routing function, refactorin
|
|||
"_arealtime", # private function for realtime API
|
||||
"acreate_realtime_client_secret",
|
||||
"arealtime_calls",
|
||||
"acreate_realtime_transcription_session",
|
||||
"_aresponses_websocket", # private function for responses WebSocket mode
|
||||
"aimage_edit",
|
||||
"agenerate_content",
|
||||
|
|
@ -427,6 +429,7 @@ async def route_request( # noqa: PLR0915 - Complex routing function, refactorin
|
|||
"adelete_run",
|
||||
"acreate_realtime_client_secret",
|
||||
"arealtime_calls",
|
||||
"acreate_realtime_transcription_session",
|
||||
]:
|
||||
# If a model is provided, get its credentials from the router
|
||||
model = data.get("model")
|
||||
|
|
|
|||
|
|
@ -1 +1,61 @@
|
|||
Abstraction / Routing logic for OpenAI's `/v1/realtime` endpoint.
|
||||
Abstraction / Routing logic for OpenAI's `/v1/realtime` endpoints.
|
||||
|
||||
## Realtime transcription (`gpt-realtime-whisper`)
|
||||
|
||||
`gpt-realtime-whisper` is the low-latency streaming speech-to-text model. It is a
|
||||
Realtime transcription session, not the file-based `/audio/transcriptions` path. Use
|
||||
the standard `gpt-4o-transcribe` / `whisper-1` models for request/response or file
|
||||
transcription; use `gpt-realtime-whisper` for live streaming transcript deltas.
|
||||
|
||||
Both OpenAI and Azure OpenAI (Microsoft Foundry) are supported. Cost is tracked by input
|
||||
audio duration (OpenAI: $0.017/minute), derived from the
|
||||
`conversation.item.input_audio_transcription.completed` usage events.
|
||||
|
||||
### WebSocket
|
||||
|
||||
Connect to the proxy realtime WebSocket with `intent=transcription`, then send a
|
||||
`session.update` configuring a transcription session:
|
||||
|
||||
```
|
||||
wss://<proxy>/v1/realtime?model=gpt-realtime-whisper&intent=transcription
|
||||
```
|
||||
|
||||
```json
|
||||
{
|
||||
"type": "session.update",
|
||||
"session": {
|
||||
"type": "transcription",
|
||||
"audio": {
|
||||
"input": {
|
||||
"format": { "type": "audio/pcm", "rate": 24000 },
|
||||
"transcription": { "model": "gpt-realtime-whisper", "language": "en" }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
Append audio with `input_audio_buffer.append`, then `input_audio_buffer.commit` (when not
|
||||
using server VAD). Listen for `conversation.item.input_audio_transcription.delta` and
|
||||
`.completed` events. The proxy does not auto-trigger `response.create` for transcription
|
||||
sessions.
|
||||
|
||||
### Ephemeral transcription session (WebRTC)
|
||||
|
||||
`POST /v1/realtime/transcription_sessions` mints an ephemeral session for browser/WebRTC
|
||||
clients. The returned `client_secret.value` is encrypted by the proxy and exchanged via
|
||||
`POST /v1/realtime/calls`.
|
||||
|
||||
```bash
|
||||
curl https://<proxy>/v1/realtime/transcription_sessions \
|
||||
-H "Authorization: Bearer $LITELLM_KEY" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{
|
||||
"input_audio_format": "pcm16",
|
||||
"input_audio_transcription": { "model": "gpt-realtime-whisper", "language": "en" }
|
||||
}'
|
||||
```
|
||||
|
||||
For Azure, route to an `azure/gpt-realtime-whisper` deployment; the proxy targets
|
||||
`/openai/realtime/transcription_sessions?api-version=...` and forwards
|
||||
`intent=transcription` on the WebSocket.
|
||||
|
|
@ -15,6 +15,7 @@ from litellm.types.realtime import (
|
|||
RealtimeExpiresAfter,
|
||||
RealtimeQueryParams,
|
||||
RealtimeSessionConfig,
|
||||
RealtimeTranscriptionSessionRequest,
|
||||
)
|
||||
from litellm.types.router import GenericLiteLLMParams
|
||||
from litellm.types.utils import LlmProviders
|
||||
|
|
@ -159,6 +160,72 @@ async def acreate_realtime_client_secret(
|
|||
)
|
||||
|
||||
|
||||
@wrapper_client
|
||||
async def acreate_realtime_transcription_session(
|
||||
model: Optional[str] = None,
|
||||
transcription_session: Optional[Dict[str, Any]] = None,
|
||||
timeout: Optional[float] = None,
|
||||
**kwargs,
|
||||
):
|
||||
"""
|
||||
Create an ephemeral transcription session via POST
|
||||
/v1/realtime/transcription_sessions.
|
||||
|
||||
``transcription_session`` is the upstream request body (input_audio_format,
|
||||
input_audio_transcription, turn_detection, …). ``model`` is a LiteLLM-only
|
||||
routing hint; the provider model lives in
|
||||
``transcription_session.input_audio_transcription.model``.
|
||||
"""
|
||||
req = RealtimeTranscriptionSessionRequest(
|
||||
model=model,
|
||||
**(transcription_session or {}),
|
||||
)
|
||||
model_name = req.resolved_model() or "gpt-realtime-whisper"
|
||||
litellm_logging_obj: LiteLLMLogging = kwargs.get("litellm_logging_obj") # type: ignore
|
||||
litellm_params = GenericLiteLLMParams(**kwargs)
|
||||
|
||||
(
|
||||
model_name,
|
||||
custom_llm_provider,
|
||||
dynamic_api_key,
|
||||
dynamic_api_base,
|
||||
) = get_llm_provider(
|
||||
model=model_name,
|
||||
api_base=litellm_params.api_base,
|
||||
api_key=litellm_params.api_key,
|
||||
)
|
||||
(
|
||||
provider_config,
|
||||
resolved_api_base,
|
||||
resolved_api_key,
|
||||
) = _get_realtime_http_provider_config(
|
||||
custom_llm_provider=custom_llm_provider,
|
||||
dynamic_api_base=dynamic_api_base,
|
||||
dynamic_api_key=dynamic_api_key,
|
||||
litellm_params=litellm_params,
|
||||
)
|
||||
litellm_logging_obj.update_from_kwargs(
|
||||
kwargs=kwargs,
|
||||
model=model_name,
|
||||
optional_params={"transcription_session": transcription_session},
|
||||
litellm_params={"api_base": resolved_api_base},
|
||||
custom_llm_provider=custom_llm_provider,
|
||||
)
|
||||
request_data = req.model_dump(exclude_none=True, exclude={"model"})
|
||||
return await base_llm_http_handler.async_realtime_transcription_session_handler(
|
||||
api_base=resolved_api_base,
|
||||
api_key=resolved_api_key,
|
||||
request_data=request_data,
|
||||
logging_obj=litellm_logging_obj,
|
||||
timeout=timeout or request_timeout,
|
||||
provider_config=provider_config,
|
||||
model=model_name,
|
||||
extra_headers=kwargs.get("extra_headers"),
|
||||
client=kwargs.get("client"),
|
||||
api_version=litellm_params.api_version,
|
||||
)
|
||||
|
||||
|
||||
@wrapper_client
|
||||
async def arealtime_calls(
|
||||
openai_ephemeral_key: str,
|
||||
|
|
@ -313,6 +380,7 @@ async def _arealtime( # noqa: PLR0915
|
|||
timeout=timeout,
|
||||
logging_obj=litellm_logging_obj,
|
||||
realtime_protocol=realtime_protocol,
|
||||
query_params=query_params,
|
||||
user_api_key_dict=kwargs.get("user_api_key_dict"),
|
||||
litellm_metadata=_build_litellm_metadata(kwargs),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -115,3 +115,40 @@ class RealtimeClientSecretResponse(BaseModel):
|
|||
expires_at: Optional[int] = None
|
||||
value: str
|
||||
session: Optional[Dict[str, Any]] = None
|
||||
|
||||
|
||||
class RealtimeTranscriptionSessionRequest(BaseModel):
|
||||
"""
|
||||
Request body for POST /v1/realtime/transcription_sessions.
|
||||
|
||||
Mirrors OpenAI's RealtimeTranscriptionSessionCreateRequest. The model used
|
||||
for routing is taken from the LiteLLM-only top-level `model` hint, falling
|
||||
back to `input_audio_transcription.model`. All other fields pass through
|
||||
unchanged to the provider.
|
||||
"""
|
||||
|
||||
model_config = {"extra": "allow"}
|
||||
|
||||
# LiteLLM-only routing hint — stripped before forwarding upstream.
|
||||
model: Optional[str] = None
|
||||
input_audio_transcription: Optional[Dict[str, Any]] = None
|
||||
|
||||
def resolved_model(self) -> Optional[str]:
|
||||
if self.model:
|
||||
return self.model
|
||||
if self.input_audio_transcription:
|
||||
return self.input_audio_transcription.get("model")
|
||||
return None
|
||||
|
||||
|
||||
class RealtimeTranscriptionSessionResponse(BaseModel):
|
||||
"""
|
||||
Response from POST /v1/realtime/transcription_sessions.
|
||||
|
||||
`client_secret.value` contains the encrypted token instead of the raw
|
||||
ephemeral key. Unknown fields pass through unchanged.
|
||||
"""
|
||||
|
||||
model_config = {"extra": "allow"}
|
||||
|
||||
client_secret: Optional[Dict[str, Any]] = None
|
||||
|
|
|
|||
|
|
@ -527,6 +527,7 @@ CallTypesLiteral = Literal[
|
|||
"acreate_skill",
|
||||
"acreate_realtime_client_secret",
|
||||
"arealtime_calls",
|
||||
"acreate_realtime_transcription_session",
|
||||
]
|
||||
|
||||
# Mapping of API routes to their corresponding call types
|
||||
|
|
|
|||
|
|
@ -4204,6 +4204,23 @@
|
|||
"/v1/audio/transcriptions"
|
||||
]
|
||||
},
|
||||
"azure/gpt-realtime-whisper": {
|
||||
"input_cost_per_second": 0.0002833333333333333,
|
||||
"litellm_provider": "azure",
|
||||
"mode": "audio_transcription",
|
||||
"source": "https://learn.microsoft.com/en-us/azure/ai-foundry/openai/concepts/gpt-realtime-whisper",
|
||||
"supported_endpoints": [
|
||||
"/v1/realtime",
|
||||
"/v1/realtime/transcription_sessions"
|
||||
],
|
||||
"supported_modalities": [
|
||||
"audio"
|
||||
],
|
||||
"supported_output_modalities": [
|
||||
"text"
|
||||
],
|
||||
"supports_audio_input": true
|
||||
},
|
||||
"azure/gpt-5.1-2025-11-13": {
|
||||
"cache_read_input_token_cost": 1.25e-07,
|
||||
"cache_read_input_token_cost_priority": 2.5e-07,
|
||||
|
|
@ -40542,6 +40559,23 @@
|
|||
"supports_system_messages": true,
|
||||
"supports_tool_choice": true
|
||||
},
|
||||
"gpt-realtime-whisper": {
|
||||
"input_cost_per_second": 0.0002833333333333333,
|
||||
"litellm_provider": "openai",
|
||||
"mode": "audio_transcription",
|
||||
"source": "https://platform.openai.com/docs/models/gpt-realtime-whisper",
|
||||
"supported_endpoints": [
|
||||
"/v1/realtime",
|
||||
"/v1/realtime/transcription_sessions"
|
||||
],
|
||||
"supported_modalities": [
|
||||
"audio"
|
||||
],
|
||||
"supported_output_modalities": [
|
||||
"text"
|
||||
],
|
||||
"supports_audio_input": true
|
||||
},
|
||||
"sora-2": {
|
||||
"litellm_provider": "openai",
|
||||
"mode": "video_generation",
|
||||
|
|
|
|||
|
|
@ -521,6 +521,117 @@ async def test_transcription_captured_in_backend_to_client():
|
|||
assert logging_obj.model_call_details["messages"] == streaming.input_messages
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_transcription_session_captures_usage_and_skips_response_create():
|
||||
"""
|
||||
For a transcription-only session (session.type == "transcription", e.g.
|
||||
gpt-realtime-whisper), the completed event's audio-duration usage must be
|
||||
captured for cost and response.create must NOT be sent to the backend.
|
||||
"""
|
||||
client_ws = MagicMock()
|
||||
client_ws.send_text = AsyncMock()
|
||||
|
||||
session_created = json.dumps(
|
||||
{
|
||||
"type": "session.created",
|
||||
"session": {
|
||||
"type": "transcription",
|
||||
"audio": {
|
||||
"input": {"transcription": {"model": "gpt-realtime-whisper"}}
|
||||
},
|
||||
},
|
||||
}
|
||||
).encode()
|
||||
completed = json.dumps(
|
||||
{
|
||||
"type": "conversation.item.input_audio_transcription.completed",
|
||||
"transcript": "hello world",
|
||||
"item_id": "item_1",
|
||||
"usage": {"type": "duration", "seconds": 12.0},
|
||||
}
|
||||
).encode()
|
||||
|
||||
backend_ws = MagicMock()
|
||||
backend_ws.recv = AsyncMock(
|
||||
side_effect=[session_created, completed, ConnectionClosed(None, None)]
|
||||
)
|
||||
backend_ws.send = AsyncMock()
|
||||
|
||||
logging_obj = MagicMock()
|
||||
logging_obj.model_call_details = {}
|
||||
logging_obj.async_success_handler = AsyncMock()
|
||||
logging_obj.success_handler = MagicMock()
|
||||
|
||||
streaming = RealTimeStreaming(client_ws, backend_ws, logging_obj)
|
||||
await streaming.backend_to_client_send_messages()
|
||||
|
||||
assert streaming._is_transcription_session is True
|
||||
|
||||
captured = [
|
||||
m
|
||||
for m in streaming.messages
|
||||
if m.get("type") == "conversation.item.input_audio_transcription.completed"
|
||||
]
|
||||
assert len(captured) == 1, "completed usage event must be captured for cost"
|
||||
assert captured[0]["usage"]["seconds"] == 12.0
|
||||
|
||||
# Transcript still forwarded to the client.
|
||||
client_ws.send_text.assert_any_call(completed.decode())
|
||||
|
||||
# No response.create — transcription sessions have no assistant turn.
|
||||
sent_to_backend = [
|
||||
json.loads(c.args[0]) for c in backend_ws.send.call_args_list if c.args
|
||||
]
|
||||
assert all(
|
||||
e.get("type") != "response.create" for e in sent_to_backend
|
||||
), f"transcription session must not trigger response.create, got: {sent_to_backend}"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_non_transcription_completed_event_still_triggers_response_create():
|
||||
"""
|
||||
Regression guard: a normal (non-transcription) session with no guardrails must
|
||||
keep triggering response.create on a completed transcription event.
|
||||
"""
|
||||
client_ws = MagicMock()
|
||||
client_ws.send_text = AsyncMock()
|
||||
|
||||
completed = json.dumps(
|
||||
{
|
||||
"type": "conversation.item.input_audio_transcription.completed",
|
||||
"transcript": "hi",
|
||||
"item_id": "item_1",
|
||||
}
|
||||
).encode()
|
||||
|
||||
backend_ws = MagicMock()
|
||||
backend_ws.recv = AsyncMock(side_effect=[completed, ConnectionClosed(None, None)])
|
||||
backend_ws.send = AsyncMock()
|
||||
|
||||
logging_obj = MagicMock()
|
||||
logging_obj.async_success_handler = AsyncMock()
|
||||
logging_obj.success_handler = MagicMock()
|
||||
|
||||
streaming = RealTimeStreaming(client_ws, backend_ws, logging_obj)
|
||||
await streaming.backend_to_client_send_messages()
|
||||
|
||||
assert streaming._is_transcription_session is False
|
||||
sent_to_backend = [
|
||||
json.loads(c.args[0]) for c in backend_ws.send.call_args_list if c.args
|
||||
]
|
||||
assert any(e.get("type") == "response.create" for e in sent_to_backend)
|
||||
|
||||
|
||||
def test_client_session_update_marks_transcription_session():
|
||||
"""A client session.update with type=transcription flags the session."""
|
||||
streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock())
|
||||
assert streaming._is_transcription_session is False
|
||||
streaming._collect_user_input_from_client_event(
|
||||
json.dumps({"type": "session.update", "session": {"type": "transcription"}})
|
||||
)
|
||||
assert streaming._is_transcription_session is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_client_ack_caches_setup_to_prevent_duplicate_session_update_setup():
|
||||
websocket = MagicMock()
|
||||
|
|
|
|||
|
|
@ -147,6 +147,78 @@ async def test_construct_url_ga_protocol():
|
|||
assert "deployment" not in url
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_construct_url_forwards_transcription_intent_ga():
|
||||
"""
|
||||
Transcription sessions connect with intent=transcription. The Azure handler
|
||||
must forward that query param so gpt-realtime-whisper opens a transcription
|
||||
session instead of a normal realtime session.
|
||||
"""
|
||||
from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime
|
||||
|
||||
handler = AzureOpenAIRealtime()
|
||||
url = handler._construct_url(
|
||||
api_base="https://my-endpoint.openai.azure.com",
|
||||
model="gpt-realtime-whisper",
|
||||
api_version="2025-04-01-preview",
|
||||
realtime_protocol="GA",
|
||||
query_params={"model": "gpt-realtime-whisper", "intent": "transcription"},
|
||||
)
|
||||
|
||||
assert "/openai/v1/realtime?" in url
|
||||
assert "intent=transcription" in url
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_construct_url_forwards_transcription_intent_beta():
|
||||
from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime
|
||||
|
||||
handler = AzureOpenAIRealtime()
|
||||
url = handler._construct_url(
|
||||
api_base="https://my-endpoint.openai.azure.com",
|
||||
model="whisper-deploy",
|
||||
api_version="2024-10-01-preview",
|
||||
query_params={"intent": "transcription"},
|
||||
)
|
||||
|
||||
assert "/openai/realtime?" in url
|
||||
assert "deployment=whisper-deploy" in url
|
||||
assert "intent=transcription" in url
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_construct_url_encodes_intent_value():
|
||||
"""A crafted intent value must be URL-encoded, not injected as raw query params."""
|
||||
from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime
|
||||
|
||||
handler = AzureOpenAIRealtime()
|
||||
url = handler._construct_url(
|
||||
api_base="https://my-endpoint.openai.azure.com",
|
||||
model="gpt-realtime-whisper",
|
||||
api_version="2025-04-01-preview",
|
||||
realtime_protocol="GA",
|
||||
query_params={"intent": "transcription&foo=bar"},
|
||||
)
|
||||
assert "intent=transcription%26foo%3Dbar" in url
|
||||
assert "&foo=bar" not in url
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_construct_url_no_intent_when_absent():
|
||||
"""No intent param leaks into the URL when not provided."""
|
||||
from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime
|
||||
|
||||
handler = AzureOpenAIRealtime()
|
||||
url = handler._construct_url(
|
||||
api_base="https://my-endpoint.openai.azure.com",
|
||||
model="gpt-4o-realtime-preview",
|
||||
api_version="2024-10-01-preview",
|
||||
realtime_protocol="GA",
|
||||
query_params={"model": "gpt-4o-realtime-preview"},
|
||||
)
|
||||
assert "intent=" not in url
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_construct_url_v1_protocol():
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -0,0 +1,174 @@
|
|||
"""
|
||||
Tests for the Realtime transcription_sessions surface used by gpt-realtime-whisper:
|
||||
- OpenAI / Azure URL construction (POST /v1/realtime/transcription_sessions)
|
||||
- RealtimeTranscriptionSessionRequest model-resolution + passthrough
|
||||
- BaseLLMHTTPHandler.async_realtime_transcription_session_handler targeting
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, os.path.abspath("../../../../.."))
|
||||
|
||||
from litellm.llms.azure.realtime.http_transformation import AzureRealtimeHTTPConfig
|
||||
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
|
||||
from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
|
||||
from litellm.llms.openai.realtime.http_transformation import OpenAIRealtimeHTTPConfig
|
||||
from litellm.types.realtime import RealtimeTranscriptionSessionRequest
|
||||
|
||||
|
||||
def test_openai_transcription_session_url():
|
||||
cfg = OpenAIRealtimeHTTPConfig()
|
||||
assert (
|
||||
cfg.get_transcription_session_url(
|
||||
api_base="https://api.openai.com", model="gpt-realtime-whisper"
|
||||
)
|
||||
== "https://api.openai.com/v1/realtime/transcription_sessions"
|
||||
)
|
||||
|
||||
|
||||
def test_openai_transcription_session_url_strips_trailing_v1():
|
||||
"""A /v1 suffix must not be duplicated in the path."""
|
||||
cfg = OpenAIRealtimeHTTPConfig()
|
||||
assert (
|
||||
cfg.get_transcription_session_url(
|
||||
api_base="https://api.openai.com/v1", model="gpt-realtime-whisper"
|
||||
)
|
||||
== "https://api.openai.com/v1/realtime/transcription_sessions"
|
||||
)
|
||||
|
||||
|
||||
def test_azure_transcription_session_url_uses_deployment_and_api_version():
|
||||
cfg = AzureRealtimeHTTPConfig()
|
||||
url = cfg.get_transcription_session_url(
|
||||
api_base="https://my.openai.azure.com",
|
||||
model="whisper-deploy",
|
||||
api_version="2025-04-01-preview",
|
||||
)
|
||||
assert (
|
||||
url
|
||||
== "https://my.openai.azure.com/openai/realtime/transcription_sessions?api-version=2025-04-01-preview"
|
||||
)
|
||||
|
||||
|
||||
def test_request_resolves_model_from_top_level_hint():
|
||||
req = RealtimeTranscriptionSessionRequest(
|
||||
model="openai/gpt-realtime-whisper",
|
||||
input_audio_transcription={"model": "gpt-realtime-whisper"},
|
||||
)
|
||||
assert req.resolved_model() == "openai/gpt-realtime-whisper"
|
||||
|
||||
|
||||
def test_request_resolves_model_from_input_audio_transcription():
|
||||
req = RealtimeTranscriptionSessionRequest(
|
||||
input_audio_transcription={"model": "gpt-realtime-whisper", "language": "en"},
|
||||
)
|
||||
assert req.resolved_model() == "gpt-realtime-whisper"
|
||||
|
||||
|
||||
def test_request_passthrough_excludes_routing_hint():
|
||||
"""Unknown fields pass through; the litellm-only `model` hint is not forwarded."""
|
||||
req = RealtimeTranscriptionSessionRequest(
|
||||
model="openai/gpt-realtime-whisper",
|
||||
input_audio_format="pcm16",
|
||||
input_audio_transcription={"model": "gpt-realtime-whisper"},
|
||||
turn_detection=None,
|
||||
)
|
||||
forwarded = req.model_dump(exclude_none=True, exclude={"model"})
|
||||
assert "model" not in forwarded
|
||||
assert forwarded["input_audio_format"] == "pcm16"
|
||||
assert forwarded["input_audio_transcription"] == {"model": "gpt-realtime-whisper"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handler_posts_to_transcription_sessions_url():
|
||||
handler = BaseLLMHTTPHandler()
|
||||
|
||||
mock_response = MagicMock(spec=httpx.Response)
|
||||
mock_client = MagicMock(spec=AsyncHTTPHandler)
|
||||
mock_client.post = AsyncMock(return_value=mock_response)
|
||||
|
||||
logging_obj = MagicMock()
|
||||
logging_obj.pre_call = MagicMock()
|
||||
|
||||
request_body = {"input_audio_transcription": {"model": "gpt-realtime-whisper"}}
|
||||
result = await handler.async_realtime_transcription_session_handler(
|
||||
api_base="https://api.openai.com",
|
||||
api_key="sk-test",
|
||||
request_data=request_body,
|
||||
logging_obj=logging_obj,
|
||||
timeout=10.0,
|
||||
provider_config=OpenAIRealtimeHTTPConfig(),
|
||||
model="gpt-realtime-whisper",
|
||||
client=mock_client,
|
||||
)
|
||||
|
||||
assert result is mock_response
|
||||
_, kwargs = mock_client.post.call_args
|
||||
assert kwargs["url"] == "https://api.openai.com/v1/realtime/transcription_sessions"
|
||||
assert kwargs["json"] == request_body
|
||||
assert kwargs["headers"]["Authorization"] == "Bearer sk-test"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_client_secret_handler_still_targets_client_secrets_url():
|
||||
"""Refactor regression: the client_secrets handler must keep its own URL."""
|
||||
handler = BaseLLMHTTPHandler()
|
||||
|
||||
mock_response = MagicMock(spec=httpx.Response)
|
||||
mock_client = MagicMock(spec=AsyncHTTPHandler)
|
||||
mock_client.post = AsyncMock(return_value=mock_response)
|
||||
|
||||
logging_obj = MagicMock()
|
||||
logging_obj.pre_call = MagicMock()
|
||||
|
||||
await handler.async_realtime_client_secret_handler(
|
||||
api_base="https://api.openai.com",
|
||||
api_key="sk-test",
|
||||
request_data={"session": {"type": "realtime"}},
|
||||
logging_obj=logging_obj,
|
||||
timeout=10.0,
|
||||
provider_config=OpenAIRealtimeHTTPConfig(),
|
||||
model="gpt-4o-realtime-preview",
|
||||
client=mock_client,
|
||||
)
|
||||
|
||||
_, kwargs = mock_client.post.call_args
|
||||
assert kwargs["url"] == "https://api.openai.com/v1/realtime/client_secrets"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_sdk_fn_routes_openai_transcription_session(monkeypatch):
|
||||
"""
|
||||
litellm.acreate_realtime_transcription_session resolves the OpenAI provider
|
||||
from the transcription model and POSTs to the OpenAI transcription_sessions URL.
|
||||
"""
|
||||
import litellm
|
||||
|
||||
monkeypatch.setenv("OPENAI_API_KEY", "sk-unit-test")
|
||||
|
||||
mock_response = MagicMock(spec=httpx.Response)
|
||||
mock_client = MagicMock(spec=AsyncHTTPHandler)
|
||||
mock_client.post = AsyncMock(return_value=mock_response)
|
||||
|
||||
result = await litellm.acreate_realtime_transcription_session(
|
||||
model="openai/gpt-realtime-whisper",
|
||||
transcription_session={
|
||||
"input_audio_format": "pcm16",
|
||||
"input_audio_transcription": {"model": "gpt-realtime-whisper"},
|
||||
},
|
||||
client=mock_client,
|
||||
)
|
||||
|
||||
assert result is mock_response
|
||||
_, kwargs = mock_client.post.call_args
|
||||
assert kwargs["url"].endswith("/v1/realtime/transcription_sessions")
|
||||
# The litellm-only routing hint must not be forwarded upstream.
|
||||
assert "model" not in kwargs["json"]
|
||||
assert kwargs["json"]["input_audio_transcription"] == {
|
||||
"model": "gpt-realtime-whisper"
|
||||
}
|
||||
|
|
@ -311,3 +311,191 @@ async def test_realtime_calls_success_with_valid_encrypted_token(
|
|||
assert response.status_code == 201
|
||||
assert response.content.startswith(b"v=0")
|
||||
assert b"application/sdp" in response.headers.get("content-type", "").encode()
|
||||
|
||||
|
||||
def test_token_payload_carries_session_type():
|
||||
"""The encrypted token records the session kind so /realtime/calls can replay it."""
|
||||
payload = _encode_realtime_token_payload(
|
||||
ephemeral_key="epk",
|
||||
model_id="gpt-realtime-whisper",
|
||||
user_id=None,
|
||||
team_id=None,
|
||||
expires_at=None,
|
||||
session_type="transcription",
|
||||
)
|
||||
decoded = _decode_realtime_token_payload(payload)
|
||||
assert decoded is not None
|
||||
assert decoded["session_type"] == "transcription"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_realtime_calls_replays_transcription_session_type(
|
||||
proxy_app,
|
||||
mock_add_litellm_data,
|
||||
mock_pre_call_hook,
|
||||
):
|
||||
"""
|
||||
A token minted for a transcription session must drive /realtime/calls to send
|
||||
session.type == "transcription" upstream, not the default "realtime".
|
||||
"""
|
||||
captured = {}
|
||||
|
||||
async def _capturing_route(*args, **kwargs):
|
||||
captured["session"] = kwargs.get("data", {}).get("session")
|
||||
|
||||
async def _inner():
|
||||
resp = MagicMock(spec=httpx.Response)
|
||||
resp.status_code = 201
|
||||
resp.content = b"v=0\r\n"
|
||||
resp.headers = {"content-type": "application/sdp"}
|
||||
return resp
|
||||
|
||||
return _inner()
|
||||
|
||||
token_payload = _encode_realtime_token_payload(
|
||||
ephemeral_key="epk",
|
||||
model_id="gpt-realtime-whisper",
|
||||
user_id=None,
|
||||
team_id=None,
|
||||
expires_at=int(time.time()) + 3600,
|
||||
session_type="transcription",
|
||||
)
|
||||
encrypted_token = encrypt_value_helper(token_payload)
|
||||
|
||||
client = TestClient(proxy_app)
|
||||
with (
|
||||
patch(
|
||||
"litellm.proxy.proxy_server.route_request",
|
||||
side_effect=_capturing_route,
|
||||
),
|
||||
patch(
|
||||
"litellm.proxy.proxy_server.add_litellm_data_to_request",
|
||||
side_effect=mock_add_litellm_data,
|
||||
),
|
||||
patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_logging,
|
||||
):
|
||||
mock_logging.pre_call_hook = AsyncMock(side_effect=mock_pre_call_hook)
|
||||
mock_logging.post_call_failure_hook = AsyncMock()
|
||||
|
||||
client.post(
|
||||
"/v1/realtime/calls",
|
||||
headers={"Authorization": f"Bearer {encrypted_token}"},
|
||||
content=b"v=0\r\n",
|
||||
)
|
||||
|
||||
assert captured["session"]["type"] == "transcription"
|
||||
|
||||
|
||||
# --- transcription_sessions endpoint ---
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_route_request_transcription_sessions():
|
||||
"""Mock route_request to return a fake transcription_sessions upstream response."""
|
||||
future_expires_at = int(time.time()) + 3600
|
||||
body = {
|
||||
"id": "sess_abc",
|
||||
"object": "realtime.transcription_session",
|
||||
"client_secret": {
|
||||
"value": "upstream_ephemeral_key",
|
||||
"expires_at": future_expires_at,
|
||||
},
|
||||
}
|
||||
mock_resp = MagicMock(spec=httpx.Response)
|
||||
mock_resp.status_code = 200
|
||||
mock_resp.text = json.dumps(body)
|
||||
mock_resp.content = json.dumps(body).encode()
|
||||
mock_resp.headers = {}
|
||||
mock_resp.json.return_value = body
|
||||
|
||||
async def _mock_route(*args, **kwargs):
|
||||
async def _inner():
|
||||
return mock_resp
|
||||
|
||||
return _inner()
|
||||
|
||||
return _mock_route
|
||||
|
||||
|
||||
def test_transcription_sessions_requires_auth(proxy_app):
|
||||
"""POST /v1/realtime/transcription_sessions returns 401 without Authorization."""
|
||||
from fastapi import HTTPException
|
||||
|
||||
def _raise_401():
|
||||
raise HTTPException(status_code=401, detail="Unauthorized")
|
||||
|
||||
proxy_app.dependency_overrides[user_api_key_auth] = _raise_401
|
||||
try:
|
||||
client = TestClient(proxy_app, raise_server_exceptions=False)
|
||||
response = client.post(
|
||||
"/v1/realtime/transcription_sessions",
|
||||
json={"input_audio_transcription": {"model": "gpt-realtime-whisper"}},
|
||||
)
|
||||
assert response.status_code == 401
|
||||
finally:
|
||||
proxy_app.dependency_overrides.pop(user_api_key_auth, None)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_transcription_sessions_encrypts_client_secret(
|
||||
proxy_app,
|
||||
mock_route_request_transcription_sessions,
|
||||
mock_add_litellm_data,
|
||||
mock_pre_call_hook,
|
||||
):
|
||||
"""
|
||||
POST /v1/realtime/transcription_sessions returns 200 and the ephemeral key
|
||||
under client_secret.value must be encrypted (never the raw upstream key).
|
||||
"""
|
||||
proxy_app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(
|
||||
user_id="test-user", team_id="test-team"
|
||||
)
|
||||
captured_route_type = {}
|
||||
|
||||
async def _capturing_route(*args, **kwargs):
|
||||
captured_route_type["route_type"] = kwargs.get("route_type")
|
||||
return await mock_route_request_transcription_sessions(*args, **kwargs)
|
||||
|
||||
try:
|
||||
client = TestClient(proxy_app)
|
||||
with (
|
||||
patch(
|
||||
"litellm.proxy.proxy_server.route_request",
|
||||
side_effect=_capturing_route,
|
||||
),
|
||||
patch(
|
||||
"litellm.proxy.proxy_server.add_litellm_data_to_request",
|
||||
side_effect=mock_add_litellm_data,
|
||||
),
|
||||
patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_logging,
|
||||
):
|
||||
mock_logging.pre_call_hook = AsyncMock(side_effect=mock_pre_call_hook)
|
||||
mock_logging.post_call_failure_hook = AsyncMock()
|
||||
|
||||
response = client.post(
|
||||
"/v1/realtime/transcription_sessions",
|
||||
headers={"Authorization": "Bearer sk-test-master-key"},
|
||||
json={
|
||||
"input_audio_format": "pcm16",
|
||||
"input_audio_transcription": {"model": "gpt-realtime-whisper"},
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["client_secret"]["value"] != "upstream_ephemeral_key"
|
||||
# The encrypted value must decrypt back to a payload carrying the raw key.
|
||||
decrypted = decrypt_value_helper(
|
||||
data["client_secret"]["value"],
|
||||
key="client_secret.value",
|
||||
exception_type="debug",
|
||||
)
|
||||
assert decrypted is not None
|
||||
assert "upstream_ephemeral_key" in decrypted
|
||||
# Routed through the dedicated transcription_sessions route type.
|
||||
assert (
|
||||
captured_route_type["route_type"]
|
||||
== "acreate_realtime_transcription_session"
|
||||
)
|
||||
finally:
|
||||
proxy_app.dependency_overrides.pop(user_api_key_auth, None)
|
||||
|
|
|
|||
|
|
@ -418,9 +418,130 @@ def test_realtime_logging_object_allows_null_transcript_in_conversation_item_add
|
|||
usage=usage,
|
||||
results=results,
|
||||
)
|
||||
|
||||
assert logging_result.usage.total_tokens == 18
|
||||
assert logging_result.results[0]["item"]["content"][0]["transcript"] is None
|
||||
assert logging_result.results[0]["item"]["content"][0]["transcript"] is None
|
||||
def test_realtime_transcription_duration_cost(monkeypatch):
|
||||
"""
|
||||
gpt-realtime-whisper transcription sessions are billed by input audio duration
|
||||
($0.017/min). The .completed events carry usage {type: duration, seconds: N};
|
||||
cost must equal total_seconds * input_cost_per_second.
|
||||
"""
|
||||
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
|
||||
monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
|
||||
|
||||
from litellm.cost_calculator import RealtimeAPITokenUsageProcessor
|
||||
|
||||
results: OpenAIRealtimeStreamList = [
|
||||
{
|
||||
"type": "session.created",
|
||||
"session": {
|
||||
"type": "transcription",
|
||||
"audio": {
|
||||
"input": {"transcription": {"model": "gpt-realtime-whisper"}}
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
"type": "conversation.item.input_audio_transcription.completed",
|
||||
"transcript": "hello",
|
||||
"usage": {"type": "duration", "seconds": 60.0},
|
||||
},
|
||||
{
|
||||
"type": "conversation.item.input_audio_transcription.completed",
|
||||
"transcript": "world",
|
||||
"usage": {"type": "duration", "seconds": 30.0},
|
||||
},
|
||||
]
|
||||
|
||||
combined = RealtimeAPITokenUsageProcessor.collect_and_combine_usage_from_realtime_stream_results(
|
||||
results=results
|
||||
)
|
||||
cost = handle_realtime_stream_cost_calculation(
|
||||
results=results,
|
||||
combined_usage_object=combined,
|
||||
custom_llm_provider="openai",
|
||||
litellm_model_name="gpt-realtime-whisper",
|
||||
)
|
||||
|
||||
# 90 seconds at $0.017/minute.
|
||||
expected = 90.0 * (0.017 / 60)
|
||||
assert abs(cost - expected) < 1e-9
|
||||
assert cost > 0 # guards against the duration branch being dropped
|
||||
|
||||
|
||||
def test_realtime_transcription_duration_cost_resolves_model_from_litellm_name(
|
||||
monkeypatch,
|
||||
):
|
||||
"""When no session event carries the ASR model, the litellm_model_name is used."""
|
||||
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
|
||||
monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
|
||||
|
||||
results: OpenAIRealtimeStreamList = [
|
||||
{
|
||||
"type": "conversation.item.input_audio_transcription.completed",
|
||||
"usage": {"type": "duration", "seconds": 120.0},
|
||||
},
|
||||
]
|
||||
cost = handle_realtime_stream_cost_calculation(
|
||||
results=results,
|
||||
combined_usage_object=Usage(),
|
||||
custom_llm_provider="azure",
|
||||
litellm_model_name="azure/gpt-realtime-whisper",
|
||||
)
|
||||
assert abs(cost - 120.0 * (0.017 / 60)) < 1e-9
|
||||
|
||||
|
||||
def test_realtime_transcription_no_completed_events_is_zero(monkeypatch):
|
||||
"""A realtime stream without transcription completed events adds no extra cost."""
|
||||
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
|
||||
monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
|
||||
|
||||
from litellm.cost_calculator import handle_realtime_transcription_cost_calculation
|
||||
|
||||
results: OpenAIRealtimeStreamList = [
|
||||
{"type": "session.created", "session": {"model": "gpt-realtime-whisper"}},
|
||||
{"type": "response.done", "response": {"usage": {}}},
|
||||
]
|
||||
assert (
|
||||
handle_realtime_transcription_cost_calculation(
|
||||
results=results,
|
||||
custom_llm_provider="openai",
|
||||
litellm_model_name="gpt-realtime-whisper",
|
||||
)
|
||||
== 0.0
|
||||
)
|
||||
|
||||
|
||||
def test_realtime_transcription_token_billed_fallback(monkeypatch):
|
||||
"""
|
||||
Token-billed transcription models price by audio/text tokens. Verify the
|
||||
fallback path multiplies audio tokens by the model's audio token cost.
|
||||
"""
|
||||
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
|
||||
monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
|
||||
|
||||
from litellm.cost_calculator import _transcription_usage_cost
|
||||
|
||||
# gpt-4o-transcribe: input_cost_per_audio_token = 2.5e-06, input_cost_per_token = 2.5e-06,
|
||||
# output_cost_per_token = 1e-05
|
||||
model_info = litellm.get_model_info(
|
||||
model="gpt-4o-transcribe", custom_llm_provider="openai"
|
||||
)
|
||||
usage = {
|
||||
"type": "tokens",
|
||||
"input_tokens": 40,
|
||||
"output_tokens": 10,
|
||||
"total_tokens": 50,
|
||||
"input_token_details": {"audio_tokens": 30, "text_tokens": 10},
|
||||
}
|
||||
cost = _transcription_usage_cost(usage, model_info)
|
||||
expected = (
|
||||
30 * 2.5e-06 # audio tokens
|
||||
+ 10 * 2.5e-06 # text tokens
|
||||
+ 10 * 1e-05 # output tokens
|
||||
)
|
||||
assert abs(cost - expected) < 1e-12
|
||||
|
||||
|
||||
def test_custom_pricing_with_router_model_id():
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue