diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 17634f93d98..bfebc9b7667 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -2857,6 +2857,10 @@ class Logging(LiteLLMLoggingBaseClass): setattr(result, "usage", transformed_usage) return result + @staticmethod + def _stream_cost_source(received_result: object, assembled_response: object) -> object: + return received_result if isinstance(received_result, TranscriptionResponse) else assembled_response + def _with_client_facing_stream_model( self, response: ModelResponse | TextCompletionResponse | ResponsesAPIResponse | InteractionsAPIResponse, @@ -3229,6 +3233,7 @@ class Logging(LiteLLMLoggingBaseClass): verbose_logger.debug("Logging Details LiteLLM-Success Call: Cache_hit=%s", cache_hit) if not self.should_run_logging(event_type="sync_success"): # prevent double logging return + received_result: Final = result start_time, end_time, result, request_logging_payload = self._prepare_success_logging( start_time=start_time, end_time=end_time, @@ -3264,7 +3269,7 @@ class Logging(LiteLLMLoggingBaseClass): self.model_call_details["complete_streaming_response"] = complete_streaming_response self._surface_response_headers_from_result(complete_streaming_response) self.model_call_details["response_cost"] = self.response_cost_calculator( - result=complete_streaming_response + result=self._stream_cost_source(received_result, complete_streaming_response) ) self._add_continued_partial_spend(complete_streaming_response) self._merge_hidden_params_from_response_into_metadata(complete_streaming_response) @@ -3802,6 +3807,7 @@ class Logging(LiteLLMLoggingBaseClass): kwargs=self.model_call_details, messages=self.model_call_details.get("messages") ) ) + received_result: Final = result start_time, end_time, result = self._success_handler_helper_fn( start_time=start_time, end_time=end_time, @@ -3843,7 +3849,7 @@ class Logging(LiteLLMLoggingBaseClass): get_base_model_from_metadata(model_call_details=self.model_call_details) # base_model defaults to None if not set on model_info self.model_call_details["response_cost"] = self.response_cost_calculator( - result=complete_streaming_response + result=self._stream_cost_source(received_result, complete_streaming_response) ) verbose_logger.debug("Model=%s; cost=%s", self.model, self.model_call_details["response_cost"]) diff --git a/tests/unit/llms/openai/transcriptions/test_gpt_transcribe.py b/tests/unit/llms/openai/transcriptions/test_gpt_transcribe.py index dc8feac29e9..10c145ba6df 100644 --- a/tests/unit/llms/openai/transcriptions/test_gpt_transcribe.py +++ b/tests/unit/llms/openai/transcriptions/test_gpt_transcribe.py @@ -362,3 +362,164 @@ async def test_azure_gpt_transcribe_sends_language_hints_in_sdk_extra_body(): assert response.text == "hello" await client.close() + + +def _one_second_wav() -> io.BytesIO: + audio_file: Final = io.BytesIO() + with wave.open(audio_file, "wb") as wav_file: + wav_file.setnchannels(1) + wav_file.setsampwidth(2) + wav_file.setframerate(16000) + wav_file.writeframes(b"\x00\x00" * 16000) + audio_file.name = "sample.wav" + audio_file.seek(0) + return audio_file + + +class _SuccessCostRecorder(CustomLogger): + def __init__(self) -> None: + super().__init__() + self.async_costs: Final[asyncio.Queue[float | None]] = asyncio.Queue() + self.sync_costs: Final[queue.Queue[float | None]] = queue.Queue() + + async def async_log_success_event(self, kwargs, response_obj, start_time, end_time) -> None: + await self.async_costs.put(kwargs["standard_logging_object"]["response_cost"]) + + def log_success_event(self, kwargs, response_obj, start_time, end_time) -> None: + self.sync_costs.put(kwargs["standard_logging_object"]["response_cost"]) + + +@pytest.mark.asyncio +async def test_atranscription_bills_the_transcript_when_the_model_ignores_stream( + local_model_cost_map: None, monkeypatch: pytest.MonkeyPatch +) -> None: + async def send_response(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"text": "hello world"}) + + recorder: Final = _SuccessCostRecorder() + monkeypatch.setattr(litellm, "callbacks", [recorder]) + client: Final = AsyncOpenAI( + api_key="sk-test", + base_url="https://example.com/v1", + http_client=httpx.AsyncClient(transport=httpx.MockTransport(send_response)), + ) + await litellm.atranscription(model="openai/whisper-1", file=_one_second_wav(), client=client) + unstreamed_cost: Final = await asyncio.wait_for(recorder.async_costs.get(), timeout=5) + ignored_stream: Final = await litellm.atranscription( + model="openai/whisper-1", file=_one_second_wav(), stream=True, client=client + ) + ignored_stream_cost: Final = await asyncio.wait_for(recorder.async_costs.get(), timeout=5) + await client.close() + + assert isinstance(ignored_stream, TranscriptionResponse) + assert ignored_stream.text == "hello world" + assert unstreamed_cost is not None + assert ignored_stream_cost == unstreamed_cost > 0 + + +def test_azure_transcription_bills_the_transcript_when_the_deployment_ignores_stream( + local_model_cost_map: None, monkeypatch: pytest.MonkeyPatch +) -> None: + def send_response(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"text": "hello world"}) + + recorder: Final = _SuccessCostRecorder() + monkeypatch.setattr(litellm, "callbacks", [recorder]) + client: Final = AzureOpenAI( + api_key="azure-test-key", + azure_endpoint="https://example.openai.azure.com", + api_version=litellm.AZURE_DEFAULT_API_VERSION, + http_client=httpx.Client(transport=httpx.MockTransport(send_response)), + ) + azure_params: Final = { + "client": client, + "api_key": "azure-test-key", + "api_base": "https://example.openai.azure.com", + "api_version": litellm.AZURE_DEFAULT_API_VERSION, + } + litellm.transcription(model="azure/whisper-1", file=_one_second_wav(), **azure_params) + unstreamed_cost: Final = recorder.sync_costs.get(timeout=5) + ignored_stream: Final = litellm.transcription( + model="azure/whisper-1", file=_one_second_wav(), stream=True, **azure_params + ) + ignored_stream_cost: Final = recorder.sync_costs.get(timeout=5) + client.close() + + assert isinstance(ignored_stream, TranscriptionResponse) + assert ignored_stream.text == "hello world" + assert unstreamed_cost is not None + assert ignored_stream_cost == unstreamed_cost > 0 + + +def _duration_priced_transcript(streamed: bool) -> httpx.Response: + usage: Final = {"type": "duration", "seconds": 2} + if not streamed: + return httpx.Response(200, json={"text": "hello world", "usage": usage}) + events: Final = ( + {"type": "transcript.text.delta", "delta": "hello world"}, + {"type": "transcript.text.done", "text": "hello world", "usage": usage}, + ) + return httpx.Response( + 200, + content="".join(f"data: {json.dumps(event)}\n\n" for event in events).encode(), + headers={"content-type": "text/event-stream"}, + ) + + +def _asks_for_stream(request_body: bytes) -> bool: + return b'name="stream"\r\n\r\ntrue' in request_body + + +@pytest.mark.asyncio +async def test_atranscription_stream_bills_the_provider_duration_like_the_unstreamed_call( + local_model_cost_map: None, monkeypatch: pytest.MonkeyPatch +) -> None: + async def send_response(request: httpx.Request) -> httpx.Response: + return _duration_priced_transcript(streamed=_asks_for_stream(await request.aread())) + + recorder: Final = _SuccessCostRecorder() + monkeypatch.setattr(litellm, "callbacks", [recorder]) + client: Final = AsyncOpenAI( + api_key="sk-test", + base_url="https://example.com/v1", + http_client=httpx.AsyncClient(transport=httpx.MockTransport(send_response)), + ) + await litellm.atranscription(model="openai/gpt-transcribe", file=_one_second_wav(), client=client) + unstreamed_cost: Final = await asyncio.wait_for(recorder.async_costs.get(), timeout=5) + stream: Final = await litellm.atranscription( + model="openai/gpt-transcribe", file=_one_second_wav(), stream=True, client=client + ) + events: Final = [event.type async for event in stream] + streamed_cost: Final = await asyncio.wait_for(recorder.async_costs.get(), timeout=5) + await client.close() + + assert events == ["transcript.text.delta", "transcript.text.done"] + assert unstreamed_cost is not None + assert streamed_cost == unstreamed_cost > 0 + + +def test_transcription_stream_bills_the_provider_duration_like_the_unstreamed_call( + local_model_cost_map: None, monkeypatch: pytest.MonkeyPatch +) -> None: + def send_response(request: httpx.Request) -> httpx.Response: + return _duration_priced_transcript(streamed=_asks_for_stream(request.read())) + + recorder: Final = _SuccessCostRecorder() + monkeypatch.setattr(litellm, "callbacks", [recorder]) + client: Final = OpenAI( + api_key="sk-test", + base_url="https://example.com/v1", + http_client=httpx.Client(transport=httpx.MockTransport(send_response)), + ) + litellm.transcription(model="openai/gpt-transcribe", file=_one_second_wav(), client=client) + unstreamed_cost: Final = recorder.sync_costs.get(timeout=5) + stream: Final = litellm.transcription( + model="openai/gpt-transcribe", file=_one_second_wav(), stream=True, client=client + ) + events: Final = [event.type for event in stream] + streamed_cost: Final = recorder.sync_costs.get(timeout=5) + client.close() + + assert events == ["transcript.text.delta", "transcript.text.done"] + assert unstreamed_cost is not None + assert streamed_cost == unstreamed_cost > 0