diff --git a/litellm/completion_extras/litellm_responses_transformation/transformation.py b/litellm/completion_extras/litellm_responses_transformation/transformation.py index 1b976f5a48b..3cf5d36b834 100644 --- a/litellm/completion_extras/litellm_responses_transformation/transformation.py +++ b/litellm/completion_extras/litellm_responses_transformation/transformation.py @@ -188,7 +188,7 @@ def _reasoning_items_from_output_items(output_items: Sequence[object]) -> tuple[ def _as_chat_reasoning_items( - reasoning_items: Sequence[_BuiltReasoningItem], + reasoning_items: Sequence[_BuiltReasoningItem | ChatCompletionReasoningItem], ) -> list[ChatCompletionReasoningItem] | None: if not reasoning_items: return None @@ -782,7 +782,32 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): else: pass # don't fail request if item in list is not supported - # If we accumulated tool calls, create a single choice with all of them + if accumulated_tool_calls and choices: + last_choice: Final = choices[-1] + last_reasoning_content: Final = getattr(last_choice.message, "reasoning_content", None) + last_reasoning_items: Final = getattr(last_choice.message, "reasoning_items", None) + merged_reasoning_content: Final = ( + " ".join(value for value in (last_reasoning_content, reasoning_content) if value) or None + ) + merged_reasoning_items: Final = _as_chat_reasoning_items( + ( + *(last_reasoning_items or ()), + *(() if pending_reasoning_item is None else (pending_reasoning_item,)), + ) + ) + merged_message: Final = Message( + role=last_choice.message.role, + content=last_choice.message.content, + annotations=getattr(last_choice.message, "annotations", None), + tool_calls=accumulated_tool_calls, + reasoning_content=merged_reasoning_content, + reasoning_items=merged_reasoning_items, + ) + return [ + *choices[:-1], + Choices(message=merged_message, finish_reason="tool_calls", index=last_choice.index), + ] + if accumulated_tool_calls: msg = Message( content=None, diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index ba8addbaaa0..d5bbf180e97 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -610,6 +610,9 @@ class Logging(LiteLLMLoggingBaseClass): self.truncated_messages_for_logging: str | list | dict | None = None # mutable-ok: logged messages shape ## TIME TO FIRST TOKEN LOGGING ## self.completion_start_time: datetime.datetime | None = None + # The model the proxy shows the client on streamed chunks. The logged streamed response carries it + # once that response is priced, the same way a non-streamed response is logged + self.client_facing_stream_model: str | None = None self._llm_caching_handler: LLMCachingHandler | None = None # INITIAL LITELLM_PARAMS @@ -2279,6 +2282,15 @@ class Logging(LiteLLMLoggingBaseClass): setattr(result, "usage", transformed_usage) return result + def _with_client_facing_stream_model( + self, + response: ModelResponse | TextCompletionResponse | ResponsesAPIResponse | InteractionsAPIResponse, + ) -> ModelResponse | TextCompletionResponse | ResponsesAPIResponse | InteractionsAPIResponse: + model: Final = self.client_facing_stream_model + if model is None or getattr(response, "model", None) in (None, model): + return response + return response.model_copy(update={"model": model}) + def _success_handler_helper_fn( self, result=None, @@ -2592,9 +2604,11 @@ class Logging(LiteLLMLoggingBaseClass): result=complete_streaming_response ) self._merge_hidden_params_from_response_into_metadata(complete_streaming_response) + logged_streaming_response: Final = self._with_client_facing_stream_model(complete_streaming_response) + self.model_call_details["complete_streaming_response"] = logged_streaming_response ## STANDARDIZED LOGGING PAYLOAD self.model_call_details["standard_logging_object"] = self._build_standard_logging_payload( - complete_streaming_response, start_time, end_time + logged_streaming_response, start_time, end_time ) standard_logging_payload: Final[StandardLoggingPayload | None] = self.model_call_details.get( "standard_logging_object" @@ -3130,10 +3144,13 @@ class Logging(LiteLLMLoggingBaseClass): await self._prepare_baseline_cache_estimate(complete_streaming_response) + logged_streaming_response: Final = self._with_client_facing_stream_model(complete_streaming_response) + self.model_call_details["async_complete_streaming_response"] = logged_streaming_response + ## STANDARDIZED LOGGING PAYLOAD try: self.model_call_details["standard_logging_object"] = self._build_standard_logging_payload( - complete_streaming_response, start_time, end_time + logged_streaming_response, start_time, end_time ) except Exception: # noqa: BLE001 # payload build must never block later callbacks (slot release) verbose_logger.exception( diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 0831bb68309..63dbab23b54 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -8951,12 +8951,17 @@ def _restamp_streaming_chunk_model( ) model_mismatch_logged = True + # The streaming wrapper keeps these same chunk objects to assemble the response it + # prices, so stamp a copy for the client and leave the provider's model for pricing. + # The logging object stamps the same model on the assembled response after pricing it. + logging_obj: Final = request_data.get("litellm_logging_obj") + if isinstance(logging_obj, LiteLLMLoggingObj): + logging_obj.client_facing_stream_model = target_model if isinstance(chunk, dict): - chunk["model"] = target_model - return chunk, model_mismatch_logged + return {**chunk, "model": target_model}, model_mismatch_logged try: - chunk.model = target_model + return chunk.model_copy(update={"model": target_model}), model_mismatch_logged except Exception as e: verbose_proxy_logger.error( "litellm_call_id=%s: failed to override chunk.model=%r on chunk_type=%s. error=%s", diff --git a/tests/integration/contracts.json b/tests/integration/contracts.json index 67c7df8ee7e..95f1f4b8d4c 100644 --- a/tests/integration/contracts.json +++ b/tests/integration/contracts.json @@ -1599,6 +1599,12 @@ ], "tests/integration/spend/test_daily_activity_key_owner_traffic.py::test_owner_is_reported_on_every_route_while_a_burst_of_requests_waits_on_the_provider": [ "quota_management.spend_tracking.key_attribution.every_route_under_provider_burst" + ], + "tests/integration/spend/test_stream_alias_billing.py::test_streamed_alias_matching_a_capability_rule_bills_the_deployment_price[vertex-mock-response]": [ + "quota_management.spend_tracking.streamed_alias_bills_deployment_price" + ], + "tests/integration/spend/test_stream_alias_billing.py::test_streamed_alias_matching_a_capability_rule_bills_the_deployment_price[anthropic-upstream]": [ + "quota_management.spend_tracking.streamed_alias_bills_deployment_price" ] }, "browser": { diff --git a/tests/integration/spend/test_stream_alias_billing.py b/tests/integration/spend/test_stream_alias_billing.py new file mode 100644 index 00000000000..72cf2d3da2c --- /dev/null +++ b/tests/integration/spend/test_stream_alias_billing.py @@ -0,0 +1,191 @@ +"""A streamed alias never replaces the deployment's model for pricing (LIT-9065). + +The proxy shows the client's alias on every streamed chunk, but the chunks kept for end-of-stream cost calculation +keep the deployment's model. "claude-opus-4.8-" is no cost-map key and only matches the claude capability +rules, whose model info carries no prices, so a stream through that alias must bill exactly what the plain alias +"integration-" bills at the same deployment rates, and the client must still see the alias it asked for +""" + +import json +from collections.abc import Callable +from hashlib import sha256 +from typing import Final +from uuid import uuid4 + +import pytest +from integration._support.client import ( + Gateway, + Scenario, + eventually, + object_value, + string_value, +) +from integration._support.database import read_rows +from integration._support.wire import Reply, Request, wire_server +from pydantic import JsonValue + + +def _sse_event(name: str, payload: dict[str, JsonValue]) -> bytes: + return f"event: {name}\ndata: {json.dumps(payload, separators=(',', ':'))}\n\n".encode() + + +def _anthropic_reply(request: Request) -> Reply: + assert request.target.endswith("/v1/messages"), request.target + body: Final = json.loads(request.body) + assert body["model"] == "claude-opus-4-8", body + if body.get("stream") is not True: + return Reply( + body=json.dumps( + { + "id": f"msg_{uuid4().hex[:12]}", + "type": "message", + "role": "assistant", + "model": "claude-opus-4-8", + "content": [{"type": "text", "text": "hi"}], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 30, "output_tokens": 40}, + } + ).encode() + ) + return Reply( + content_type="text/event-stream", + chunks=( + _sse_event( + "message_start", + { + "type": "message_start", + "message": { + "id": f"msg_{uuid4().hex[:12]}", + "type": "message", + "role": "assistant", + "model": "claude-opus-4-8", + "content": [], + "stop_reason": None, + "stop_sequence": None, + "usage": {"input_tokens": 30, "output_tokens": 1}, + }, + }, + ), + _sse_event( + "content_block_start", + {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}, + ), + _sse_event( + "content_block_delta", + {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "hi"}}, + ), + _sse_event("content_block_stop", {"type": "content_block_stop", "index": 0}), + _sse_event( + "message_delta", + { + "type": "message_delta", + "delta": {"stop_reason": "end_turn", "stop_sequence": None}, + "usage": {"output_tokens": 40}, + }, + ), + _sse_event("message_stop", {"type": "message_stop"}), + ), + ) + + +def _deployment( + scenario: Scenario, + model_name: str, + litellm_params: dict[str, JsonValue], + model_info: dict[str, JsonValue] | None = None, +) -> str: + created: Final = scenario.gateway.post( + "/model/new", {"model_name": model_name, "litellm_params": litellm_params, "model_info": model_info or {}} + ) + identity: Final = string_value(object_value(created["model_info"])["id"]) + scenario.cleanups.callback(scenario.delete_model, identity) + return model_name + + +def _streamed_spend(gateway: Gateway, scenario: Scenario, model: str, content: str) -> dict[str, JsonValue]: + key: Final = scenario.key(models=[model]) + response: Final = gateway.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "messages": [{"role": "user", "content": content}], + "stream": True, + "stream_options": {"include_usage": True}, + }, + key=key, + ) + assert response.status_code == 200, response.text + chunks: Final = tuple( + json.loads(line.removeprefix("data: ")) + for line in response.text.splitlines() + if line.startswith("data: ") and line != "data: [DONE]" + ) + assert chunks and {chunk["model"] for chunk in chunks} == {model}, response.text + rows: Final = eventually( + lambda: read_rows( + 'SELECT spend, prompt_tokens, completion_tokens FROM "LiteLLM_SpendLogs" WHERE api_key=%s', + (sha256(key.encode()).hexdigest(),), + ), + lambda values: len(values) == 1, + seconds=70, + ) + return rows[0] + + +def _listed_deployments(gateway: Gateway, model_name: str) -> tuple[dict[str, JsonValue], ...]: + entries: Final = gateway.get("/model/info")["data"] + assert isinstance(entries, list) + return tuple(object_value(entry) for entry in entries if object_value(entry)["model_name"] == model_name) + + +def _deployment_pricing(gateway: Gateway, model_name: str) -> dict[str, JsonValue]: + listed: Final = eventually(lambda: _listed_deployments(gateway, model_name), lambda found: len(found) == 1) + return object_value(listed[0]["model_info"]) + + +_BACKENDS: Final = ( + pytest.param( + lambda _: {"model": "vertex_ai/claude-opus-4-8@default", "mock_response": "hi"}, + id="vertex-mock-response", + ), + pytest.param( + lambda wire_url: { + "model": "anthropic/claude-opus-4-8", + "api_key": "integration-provider-key", + "api_base": wire_url, + }, + id="anthropic-upstream", + ), +) + + +@pytest.mark.covers("quota_management.spend_tracking.streamed_alias_bills_deployment_price") +@pytest.mark.parametrize("litellm_params", _BACKENDS) +@pytest.mark.timeout(180) +def test_streamed_alias_matching_a_capability_rule_bills_the_deployment_price( + gateway: Gateway, litellm_params: Callable[[str], dict[str, JsonValue]] +) -> None: + with wire_server(_anthropic_reply) as wire, gateway.scenario() as scenario: + content: Final = f"alias billing {uuid4().hex}" + plain_alias: Final = f"integration-{uuid4().hex}" + rule_alias: Final = f"claude-opus-4.8-{uuid4().int % 10**8:08d}" + exact_row: Final = _streamed_spend( + gateway, scenario, _deployment(scenario, plain_alias, litellm_params(wire.url)), content + ) + alias_row: Final = _streamed_spend( + gateway, scenario, _deployment(scenario, rule_alias, litellm_params(wire.url)), content + ) + + for model_name, row in ((plain_alias, exact_row), (rule_alias, alias_row)): + pricing: Final = _deployment_pricing(gateway, model_name) + input_rate: Final = float(str(pricing["input_cost_per_token"])) + output_rate: Final = float(str(pricing["output_cost_per_token"])) + uplift: Final = float(str(pricing["regional_endpoint_uplift_multiplier"] or 1)) + assert input_rate > 0 and output_rate > 0, pricing + assert float(str(row["spend"])) == pytest.approx( + uplift + * (float(str(row["prompt_tokens"])) * input_rate + float(str(row["completion_tokens"])) * output_rate) + ), (model_name, row, pricing) + diff --git a/tests/test_litellm/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py b/tests/test_litellm/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py index c326ad4a0f7..c4d390e50d2 100644 --- a/tests/test_litellm/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py +++ b/tests/test_litellm/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py @@ -7,6 +7,15 @@ from unittest.mock import ANY, MagicMock, Mock, patch import httpx import pytest +from openai.types.responses import ( + ResponseFunctionToolCall, + ResponseOutputMessage, + ResponseOutputText, +) +from openai.types.responses.response_reasoning_item import ( + ResponseReasoningItem, + Summary, +) import litellm from litellm.completion_extras.litellm_responses_transformation.transformation import ( @@ -3275,6 +3284,148 @@ def test_convert_response_output_generic_pydantic_message_item(): assert choices[0].finish_reason == "stop" +def test_convert_response_output_merges_message_reasoning_and_function_call() -> None: + message: Final = ResponseOutputMessage( + id="msg_weather", + content=[ + ResponseOutputText( + annotations=[ + { + "type": "url_citation", + "start_index": 0, + "end_index": 5, + "title": "Forecast", + "url": "https://example.com/forecast", + } + ], + text="Sunny.", + type="output_text", + logprobs=[], + ) + ], + role="assistant", + status="completed", + type="message", + ) + reasoning: Final = ResponseReasoningItem( + id="rs_before", + summary=[Summary(type="summary_text", text="Checking the forecast.")], + type="reasoning", + content=None, + encrypted_content=None, + status=None, + ) + pending_reasoning: Final = ResponseReasoningItem( + id="rs_after", + summary=[Summary(type="summary_text", text="The location is Paris.")], + type="reasoning", + content=None, + encrypted_content=None, + status=None, + ) + function_call: Final = ResponseFunctionToolCall( + id="fc_1", + type="function_call", + status="completed", + arguments='{"city":"Paris"}', + call_id="call_1", + name="get_weather", + ) + + message_and_call: Final = LiteLLMResponsesTransformationHandler._convert_response_output_to_choices( + (message, function_call) + ) + assert len(message_and_call) == 1 + assert message_and_call[0].index == 0 + assert message_and_call[0].finish_reason == "tool_calls" + assert message_and_call[0].message.role == "assistant" + assert message_and_call[0].message.content == "Sunny." + assert message_and_call[0].message.annotations == [ + { + "type": "url_citation", + "start_index": 0, + "end_index": 5, + "title": "Forecast", + "url": "https://example.com/forecast", + } + ] + function_calls: Final = message_and_call[0].message.tool_calls + assert function_calls is not None + assert len(function_calls) == 1 + assert function_calls[0].function.name == "get_weather" + assert function_calls[0].function.arguments == '{"city":"Paris"}' + + reasoning_before_message: Final = LiteLLMResponsesTransformationHandler._convert_response_output_to_choices( + (reasoning, message, function_call) + ) + assert len(reasoning_before_message) == 1 + assert reasoning_before_message[0].message.reasoning_content == "Checking the forecast." + reasoning_before_items: Final = reasoning_before_message[0].message.reasoning_items + assert reasoning_before_items is not None + assert reasoning_before_items[0]["id"] == "rs_before" + + reasoning_after_message: Final = LiteLLMResponsesTransformationHandler._convert_response_output_to_choices( + (message, pending_reasoning, function_call) + ) + assert len(reasoning_after_message) == 1 + assert reasoning_after_message[0].message.reasoning_content == "The location is Paris." + reasoning_after_items: Final = reasoning_after_message[0].message.reasoning_items + assert reasoning_after_items is not None + assert reasoning_after_items[0]["id"] == "rs_after" + + merged_reasoning: Final = LiteLLMResponsesTransformationHandler._convert_response_output_to_choices( + (reasoning, message, pending_reasoning, function_call) + ) + assert len(merged_reasoning) == 1 + assert merged_reasoning[0].message.reasoning_content == "Checking the forecast. The location is Paris." + merged_reasoning_items: Final = merged_reasoning[0].message.reasoning_items + assert merged_reasoning_items is not None + assert [item["id"] for item in merged_reasoning_items] == ["rs_before", "rs_after"] + + tool_only: Final = LiteLLMResponsesTransformationHandler._convert_response_output_to_choices((function_call,)) + assert len(tool_only) == 1 + assert tool_only[0].index == 0 + assert tool_only[0].finish_reason == "tool_calls" + assert tool_only[0].message.content is None + assert tool_only[0].message.tool_calls is not None + assert len(tool_only[0].message.tool_calls) == 1 + + message_only: Final = LiteLLMResponsesTransformationHandler._convert_response_output_to_choices((message,)) + assert len(message_only) == 1 + assert message_only[0].index == 0 + assert message_only[0].finish_reason == "stop" + assert message_only[0].message.content == "Sunny." + assert message_only[0].message.tool_calls is None + + +def test_convert_response_output_merges_raw_dict_message_and_function_call() -> None: + handler: Final = LiteLLMResponsesTransformationHandler() + raw_message: Final = { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "Let me check.", "annotations": []}], + } + raw_function_call: Final = { + "type": "function_call", + "id": "fc_1", + "call_id": "call_1", + "name": "get_weather", + "arguments": '{"city":"Paris"}', + } + choices: Final = LiteLLMResponsesTransformationHandler._convert_response_output_to_choices( + (raw_message, raw_function_call), + handle_raw_dict_callback=handler._handle_raw_dict_response_item, + ) + + assert len(choices) == 1 + assert choices[0].index == 0 + assert choices[0].finish_reason == "tool_calls" + assert choices[0].message.role == "assistant" + assert choices[0].message.content == "Let me check." + assert choices[0].message.tool_calls is not None + assert len(choices[0].message.tool_calls) == 1 + + def test_convert_tools_to_responses_format_flattens_nested_custom_tool(): from litellm.completion_extras.litellm_responses_transformation.transformation import ( LiteLLMResponsesTransformationHandler, diff --git a/tests/test_litellm/litellm_core_utils/test_litellm_logging.py b/tests/test_litellm/litellm_core_utils/test_litellm_logging.py index 626a13c8061..792a7bbaa89 100644 --- a/tests/test_litellm/litellm_core_utils/test_litellm_logging.py +++ b/tests/test_litellm/litellm_core_utils/test_litellm_logging.py @@ -47,6 +47,18 @@ def logging_obj(): ) +def test_with_client_facing_stream_model_stamps_a_copy_of_the_priced_response(logging_obj): + response = ModelResponse(model="claude-opus-4-6@default") + logging_obj.client_facing_stream_model = "claude-opus-4.6" + logged = logging_obj._with_client_facing_stream_model(response) + assert (logged.model, response.model) == ("claude-opus-4.6", "claude-opus-4-6@default") + + +def test_with_client_facing_stream_model_keeps_the_response_when_the_proxy_set_no_model(logging_obj): + response = ModelResponse(model="claude-opus-4-6@default") + assert logging_obj._with_client_facing_stream_model(response) is response + + def test_get_combined_callback_list_preserves_insertion_order(logging_obj): assert logging_obj.get_combined_callback_list( dynamic_success_callbacks=["prometheus", "langfuse", "datadog", "otel", "s3"], diff --git a/tests/test_litellm/proxy/proxy_server/test_streaming_helpers.py b/tests/test_litellm/proxy/proxy_server/test_streaming_helpers.py index 86dd356e5f5..228f5a4c16d 100644 --- a/tests/test_litellm/proxy/proxy_server/test_streaming_helpers.py +++ b/tests/test_litellm/proxy/proxy_server/test_streaming_helpers.py @@ -283,8 +283,9 @@ def test_restamp_streaming_chunk_model_overrides_model_on_basemodel(): "model": new_chunk.model, "logged": logged, "same_object": new_chunk is chunk, + "original_model": chunk.model, } - assert snapshot == {"model": "gpt-4", "logged": True, "same_object": True} + assert snapshot == {"model": "gpt-4", "logged": True, "same_object": False, "original_model": "openai/internal-x"} @pytest.mark.parametrize("return_raw_model_name", [False, True]) @@ -310,8 +311,7 @@ def test_restamp_streaming_chunk_model_overrides_model_on_dict(): request_data={}, model_mismatch_logged=True, ) - assert new_chunk["model"] == "gpt-4" - assert logged is True + assert (new_chunk["model"], chunk["model"], logged) == ("gpt-4", "internal", True) def test_restamp_streaming_chunk_model_uses_fallback_model_from_metadata(): @@ -443,7 +443,7 @@ def test_restamp_streaming_chunk_model_fastest_response_preserves_model(): assert logged is False -def test_restamp_streaming_chunk_model_setattr_exception_logs_and_returns(): +def test_restamp_streaming_chunk_model_restamps_a_frozen_chunk_through_a_copy(): from pydantic import ConfigDict class FrozenChunk(_simple_chunk().__class__): @@ -462,8 +462,30 @@ def test_restamp_streaming_chunk_model_setattr_exception_logs_and_returns(): request_data={"litellm_call_id": "test-id"}, model_mismatch_logged=False, ) - assert new_chunk.model == "openai/internal-x" - assert logged is True + assert (new_chunk.model, chunk.model, logged) == ("gpt-4", "openai/internal-x", True) + + +def test_restamp_streaming_chunk_model_records_the_client_model_on_the_logging_object(): + import time + + from litellm.litellm_core_utils.litellm_logging import Logging + + logging_obj = Logging( + model="openai/internal-x", + messages=[], + stream=True, + call_type="acompletion", + start_time=time.time(), + litellm_call_id="test-id", + function_id="test-id", + ) + _restamp_streaming_chunk_model( + chunk=_simple_chunk(model="openai/internal-x"), + requested_model_from_client="gpt-4", + request_data={"litellm_call_id": "test-id", "litellm_logging_obj": logging_obj}, + model_mismatch_logged=False, + ) + assert logging_obj.client_facing_stream_model == "gpt-4" def test_format_fallback_metadata_sse_event():