scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-08 03:08:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 115

fix(gemini): bill transcribe-live sessions from streamed audio duration

Browse source 
Gemini Live sends no usageMetadata and no turnComplete for
gemini-3.5-transcribe-live sessions, so realtime spend logged as 0.0.
Attach estimated usage to the input_audio_transcription.completed event
using Google's published billing estimate (25 audio tokens/sec of input,
175 text tokens/min of output) derived from the streamed pcm16 audio
duration, gated to audio_transcription-mode models so conversational
Live models keep billing through usageMetadata. Also capture that usage
in the provider_config backend path so realtime cost calculation sees it.
This commit is contained in:
mateo-berri 2026-08-27 11:03:41 -07:00
parent e44e2fe242
commit 462942de65
5 changed files with 225 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
litellm/litellm_core_utils/realtime_streaming.py
View file

@ -955,6 +955,7 @@ class RealTimeStreaming:
transcript = event.get("transcript", "")
self._collect_user_input_from_backend_event(cast(dict, event))
self.store_message(event_str)
self._capture_transcription_usage(event)
await self._send_event_to_client(event, event_str)
blocked = await self.run_realtime_guardrails(
cast(str, transcript),

40
litellm/llms/gemini/realtime/transformation.py
View file

@ -53,6 +53,7 @@ from litellm.types.llms.vertex_ai import (
)
from litellm.types.realtime import (
ALL_DELTA_TYPES,
RealtimeInputAudioTranscriptionUsage,
RealtimeModalityResponseTransformOutput,
RealtimeResponseTransformInput,
RealtimeResponseTypedDict,
@ -95,6 +96,18 @@ def _gemini_live_speech_config(voice: object) -> Mapping[str, object] | None:
return VertexGeminiConfig()._map_audio_params({"voice": voice})
# Google bills Live transcription at an estimated 25 audio tokens/sec of input and
# 175 text tokens/min of output (ai.google.dev/gemini-api/docs/pricing).
GEMINI_LIVE_TRANSCRIBE_AUDIO_TOKENS_PER_SECOND: Final = 25
GEMINI_LIVE_TRANSCRIBE_OUTPUT_TEXT_TOKENS_PER_MINUTE: Final = 175
PCM16_INPUT_AUDIO_BYTES_PER_SECOND: Final = 48000
def _base64_decoded_byte_count(data: str) -> int:
padding: Final = 2 if data.endswith("==") else 1 if data.endswith("=") else 0
return max(len(data) * 3 // 4 - padding, 0)
class GeminiRealtimeConfig(BaseRealtimeConfig):
_TOOL_CALL_ID_TO_NAME_MAX = 256 # LRU cap for call_id→name mapping
@ -104,6 +117,7 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
# Gemini Live sometimes emits usageMetadata in a standalone frame between
# turns; buffer it here so the next response.done carries the token counts.
self._pending_usage_metadata: dict | None = None
self._unbilled_input_audio_bytes: int = 0
def is_setup_message(self, msg_obj: dict) -> bool:
return "setup" in msg_obj
@ -566,9 +580,10 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
return self._handle_conversation_item(json_message)
if msg_type == "input_audio_buffer.append":
realtime_input_dict["audio"] = HttpxBlobType(
mimeType=self.get_audio_mime_type(), data=json_message["audio"]
)
audio_b64: Final = json_message["audio"]
if isinstance(audio_b64, str):
self._unbilled_input_audio_bytes += _base64_decoded_byte_count(audio_b64)
realtime_input_dict["audio"] = HttpxBlobType(mimeType=self.get_audio_mime_type(), data=audio_b64)
realtime_input_dict = cast(
BidiGenerateContentRealtimeInput,
@ -1159,6 +1174,23 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
raise ValueError(f"Unknown openai event: {key}, value: {value}")
return openai_event
def _consume_input_transcription_usage_estimate(self, model: str) -> RealtimeInputAudioTranscriptionUsage | None:
"""Gemini Live sends no usageMetadata for transcribe sessions; estimate billing from streamed audio duration."""
if self._unbilled_input_audio_bytes <= 0 or not self._is_text_only_live_model(model):
return None
audio_seconds: Final = self._unbilled_input_audio_bytes / PCM16_INPUT_AUDIO_BYTES_PER_SECOND
self._unbilled_input_audio_bytes = 0
audio_tokens: Final = round(audio_seconds * GEMINI_LIVE_TRANSCRIBE_AUDIO_TOKENS_PER_SECOND)
output_tokens: Final = round(audio_seconds * GEMINI_LIVE_TRANSCRIBE_OUTPUT_TEXT_TOKENS_PER_MINUTE / 60)
usage: Final[RealtimeInputAudioTranscriptionUsage] = {
"type": "tokens",
"input_tokens": audio_tokens,
"output_tokens": output_tokens,
"total_tokens": audio_tokens + output_tokens,
"input_token_details": {"text_tokens": 0, "audio_tokens": audio_tokens},
}
return usage
def transform_realtime_response(
self,
message: str | bytes,
@ -1198,6 +1230,7 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
if isinstance(server_content, dict):
input_tx: Final = server_content.get("inputTranscription")
if isinstance(input_tx, dict) and input_tx.get("text"):
transcription_usage: Final = self._consume_input_transcription_usage_estimate(model)
returned_message.append(
cast(
OpenAIRealtimeEvents,
@ -1207,6 +1240,7 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
"transcript": input_tx["text"],
"item_id": f"item_{uuid.uuid4()}",
"content_index": 0,
**({} if transcription_usage is None else {"usage": transcription_usage}),
},
)
)

13
litellm/types/realtime.py
View file

@ -162,3 +162,16 @@ class RealtimeErrorDetail(TypedDict):
class RealtimeErrorEvent(TypedDict):
type: ReadOnly[Literal["error"]]
error: ReadOnly[RealtimeErrorDetail]
class RealtimeInputAudioTranscriptionUsageInputTokenDetails(TypedDict):
text_tokens: ReadOnly[int]
audio_tokens: ReadOnly[int]
class RealtimeInputAudioTranscriptionUsage(TypedDict):
type: ReadOnly[Literal["tokens"]]
input_tokens: ReadOnly[int]
output_tokens: ReadOnly[int]
total_tokens: ReadOnly[int]
input_token_details: ReadOnly[RealtimeInputAudioTranscriptionUsageInputTokenDetails]

62
tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
View file

@ -2957,3 +2957,65 @@ async def test_log_messages_routes_async_logging_through_bounded_worker():
logging_obj.success_handler.assert_not_called()
# the bare create_task path must no longer be used for success logging
mock_create_task.assert_not_called()
@pytest.mark.asyncio
async def test_provider_config_path_captures_transcription_usage():
"""A transcription.completed event with usage from the provider transform must
land in the logged messages so realtime cost calculation can bill it."""
from typing import Final
from litellm.types.realtime import RealtimeInputAudioTranscriptionUsage, RealtimeResponseTypedDict
client_ws: Final = MagicMock()
client_ws.send_text = AsyncMock()
backend_ws: Final = MagicMock()
backend_ws.send = AsyncMock()
logging_obj: Final = MagicMock()
usage: Final[RealtimeInputAudioTranscriptionUsage] = {
"type": "tokens",
"input_tokens": 50,
"output_tokens": 6,
"total_tokens": 56,
"input_token_details": {"text_tokens": 0, "audio_tokens": 50},
}
transform_output: Final[RealtimeResponseTypedDict] = {
"response": {
"type": "conversation.item.input_audio_transcription.completed",
"event_id": "event_1",
"transcript": "ahoy",
"item_id": "item_1",
"content_index": 0,
"usage": usage,
},
"current_output_item_id": None,
"current_response_id": None,
"current_delta_chunks": None,
"current_conversation_id": None,
"current_item_chunks": None,
"current_delta_type": None,
"session_configuration_request": None,
}
provider_config: Final = MagicMock()
provider_config.transform_realtime_request = MagicMock(return_value=())
provider_config.transform_realtime_response = MagicMock(return_value=transform_output)
streaming: Final = RealTimeStreaming(
client_ws,
backend_ws,
logging_obj,
provider_config=provider_config,
model="gemini-3.5-transcribe-live",
)
await streaming._handle_provider_config_message("{}")
usage_events: Final = tuple(
message
for message in streaming.messages
if isinstance(message, dict)
and message.get("type") == "conversation.item.input_audio_transcription.completed"
and message.get("usage") == usage
)
assert len(usage_events) == 1

112
tests/test_litellm/llms/gemini/realtime/test_gemini_realtime_transformation.py
View file

@ -2007,3 +2007,115 @@ def test_bare_generation_complete_without_prior_delta_is_dropped(patch_gemini_au
)
assert result["response"] == []
def _input_audio_append_message(raw_byte_count: int) -> str:
import base64
return json.dumps(
{"type": "input_audio_buffer.append", "audio": base64.b64encode(b"\x00" * raw_byte_count).decode()}
)
def test_transcribe_live_completed_event_carries_estimated_usage(patch_gemini_transcribe_live_cost_map_entry):
"""Gemini Live sends no usageMetadata for transcribe sessions, so LiteLLM bills
from streamed audio duration at Google's published estimate (25 audio tok/sec in,
175 text tok/min out): 96000 pcm16 bytes = 2s at 24kHz -> 50 in / 6 out."""
from typing import Final
from litellm.types.llms.gemini import BidiGenerateContentServerMessage
from litellm.types.realtime import RealtimeInputAudioTranscriptionUsage, RealtimeResponseTransformInput
config: Final = GeminiRealtimeConfig()
config.transform_realtime_request(_input_audio_append_message(96000), "gemini-3.5-transcribe-live")
transcript_frame: Final[BidiGenerateContentServerMessage] = {
"serverContent": {"inputTranscription": {"text": "ahoy there"}}
}
transform_input: Final[RealtimeResponseTransformInput] = {
"session_configuration_request": None,
"current_output_item_id": None,
"current_response_id": None,
"current_conversation_id": None,
"current_delta_chunks": None,
"current_item_chunks": None,
"current_delta_type": None,
}
result: Final = config.transform_realtime_response(
json.dumps(transcript_frame),
"gemini-3.5-transcribe-live",
MagicMock(),
realtime_response_transform_input=transform_input,
)
completed: Final = tuple(
event
for event in result["response"]
if event["type"] == "conversation.item.input_audio_transcription.completed"
)
assert len(completed) == 1
assert completed[0]["transcript"] == "ahoy there"
expected_usage: Final[RealtimeInputAudioTranscriptionUsage] = {
"type": "tokens",
"input_tokens": 50,
"output_tokens": 6,
"total_tokens": 56,
"input_token_details": {"text_tokens": 0, "audio_tokens": 50},
}
assert completed[0]["usage"] == expected_usage
second: Final = config.transform_realtime_response(
json.dumps(transcript_frame),
"gemini-3.5-transcribe-live",
MagicMock(),
realtime_response_transform_input=transform_input,
)
second_completed: Final = tuple(
event
for event in second["response"]
if event["type"] == "conversation.item.input_audio_transcription.completed"
)
assert len(second_completed) == 1
assert "usage" not in second_completed[0]
def test_non_transcription_live_model_completed_event_has_no_usage(patch_gemini_audio_cost_map_entries):
"""Conversational Live models get their audio tokens from usageMetadata via
response.done; attaching estimated usage to their transcription events would
double-bill, so the estimate is gated to audio_transcription-mode models."""
from typing import Final
from litellm.types.llms.gemini import BidiGenerateContentServerMessage
from litellm.types.realtime import RealtimeResponseTransformInput
config: Final = GeminiRealtimeConfig()
config.transform_realtime_request(_input_audio_append_message(96000), "gemini-3.1-flash-live-preview")
transcript_frame: Final[BidiGenerateContentServerMessage] = {
"serverContent": {"inputTranscription": {"text": "ahoy there"}}
}
transform_input: Final[RealtimeResponseTransformInput] = {
"session_configuration_request": None,
"current_output_item_id": None,
"current_response_id": None,
"current_conversation_id": None,
"current_delta_chunks": None,
"current_item_chunks": None,
"current_delta_type": None,
}
result: Final = config.transform_realtime_response(
json.dumps(transcript_frame),
"gemini-3.1-flash-live-preview",
MagicMock(),
realtime_response_transform_input=transform_input,
)
completed: Final = tuple(
event
for event in result["response"]
if event["type"] == "conversation.item.input_audio_transcription.completed"
)
assert len(completed) == 1
assert "usage" not in completed[0]