diff --git a/litellm/litellm_core_utils/llm_cost_calc/utils.py b/litellm/litellm_core_utils/llm_cost_calc/utils.py index 191231f3e6..3fd913958d 100644 --- a/litellm/litellm_core_utils/llm_cost_calc/utils.py +++ b/litellm/litellm_core_utils/llm_cost_calc/utils.py @@ -684,6 +684,9 @@ def generic_cost_per_token( # noqa: PLR0915 - cache_creation - image_tokens ) + # Clamp to zero: inconsistent streaming usage + if text_tokens < 0: + text_tokens = 0 prompt_tokens_details["text_tokens"] = text_tokens ( diff --git a/litellm/llms/bedrock/messages/invoke_transformations/anthropic_claude3_transformation.py b/litellm/llms/bedrock/messages/invoke_transformations/anthropic_claude3_transformation.py index 141d63950a..31e0e76fd9 100644 --- a/litellm/llms/bedrock/messages/invoke_transformations/anthropic_claude3_transformation.py +++ b/litellm/llms/bedrock/messages/invoke_transformations/anthropic_claude3_transformation.py @@ -591,11 +591,14 @@ class AmazonAnthropicClaudeMessagesConfig( """ Bedrock invoke does not return SSE formatted data. This function is a wrapper to ensure litellm chunks are SSE formatted. - Bedrock's Anthropic-compatible streaming puts cache usage fields - (cache_creation_input_tokens, cache_read_input_tokens) only on - message_stop, not on message_start or message_delta. Claude Code's - SDK only merges usage from message_delta, so we promote those fields - from message_stop onto message_delta before yielding. + Bedrock's Anthropic-compatible streaming usually puts cache usage fields + (cache_creation_input_tokens, cache_read_input_tokens) on message_stop. + Some deployments (including GovCloud) emit the cache breakdown only on + ``message_start.message.usage``; ``message_delta`` / ``message_stop`` then + repeat uncached ``input_tokens`` only. We promote cache fields from + ``message_stop`` onto ``message_delta``, and when those are absent we + merge them from ``message_start`` so logging/cost sees a consistent usage + object (fixes negative input costs: LIT-2411). """ from litellm.llms.anthropic.experimental_pass_through.messages.streaming_iterator import ( BaseAnthropicMessagesStreamingIterator, @@ -611,6 +614,27 @@ class AmazonAnthropicClaudeMessagesConfig( async for chunk in handler.async_sse_wrapper(patched_stream): yield chunk + @staticmethod + def _merge_message_start_cache_into_delta_usage( + delta_usage: Dict[str, Any], + start_usage: Optional[Dict[str, Any]], + ) -> None: + """ + Copy cache breakdown from message_start onto message_delta usage when + those keys are missing on the delta (GovCloud / some Bedrock streams). + """ + if not start_usage: + return + for field in ("cache_creation_input_tokens", "cache_read_input_tokens"): + if field not in delta_usage: + val = start_usage.get(field) + if val is not None: + delta_usage[field] = val + if "cache_creation" not in delta_usage: + cc = start_usage.get("cache_creation") + if cc is not None: + delta_usage["cache_creation"] = cc + @staticmethod async def _promote_message_stop_usage( completion_stream: AsyncIterator[ @@ -618,20 +642,13 @@ class AmazonAnthropicClaudeMessagesConfig( ], ) -> AsyncIterator[Union[bytes, GenericStreamingChunk, ModelResponseStream, dict]]: """ - Promote cache usage fields from message_stop onto message_delta. - - Bedrock reports input_tokens (uncached only) on message_start, and - the full breakdown (input_tokens, cache_creation_input_tokens, - cache_read_input_tokens) only on message_stop. Claude Code's SDK - merges usage from message_start and message_delta but ignores - message_stop. This method buffers message_delta and, when - message_stop arrives with cache usage, merges those fields into the - message_delta usage. input_tokens is kept as the uncached-only - count; downstream calculate_usage adds cache tokens to - prompt_tokens. + Promote cache usage fields onto message_delta from message_stop (and, + when stop lacks them, from message_start). Ensures the final usage + chunk that logging/cost sees is always self-consistent. """ _CACHE_FIELDS = ("cache_creation_input_tokens", "cache_read_input_tokens") - pending_delta = None + pending_delta: Optional[Dict[str, Any]] = None + start_usage_snapshot: Optional[Dict[str, Any]] = None async for chunk in completion_stream: if not isinstance(chunk, dict): @@ -643,8 +660,19 @@ class AmazonAnthropicClaudeMessagesConfig( chunk_type = chunk.get("type") + if chunk_type == "message_start": + msg: Dict[str, Any] = cast(Dict[str, Any], chunk.get("message") or {}) + u = msg.get("usage") + if isinstance(u, dict): + start_usage_snapshot = dict(u) + if pending_delta is not None: + yield pending_delta + pending_delta = None + yield chunk + continue + if chunk_type == "message_delta": - pending_delta = chunk + pending_delta = cast(Dict[str, Any], chunk) continue if chunk_type == "message_stop" and pending_delta is not None: @@ -661,6 +689,10 @@ class AmazonAnthropicClaudeMessagesConfig( raw_input if isinstance(raw_input, int) else 0 ) + AmazonAnthropicClaudeMessagesConfig._merge_message_start_cache_into_delta_usage( + delta_usage, start_usage_snapshot + ) + if delta_usage: pending_delta["usage"] = delta_usage # type: ignore[arg-type] @@ -676,6 +708,12 @@ class AmazonAnthropicClaudeMessagesConfig( yield chunk if pending_delta is not None: + delta_usage = dict(pending_delta.get("usage") or {}) + AmazonAnthropicClaudeMessagesConfig._merge_message_start_cache_into_delta_usage( + delta_usage, start_usage_snapshot + ) + if delta_usage: + pending_delta["usage"] = delta_usage # type: ignore[arg-type] yield pending_delta diff --git a/tests/test_litellm/llms/bedrock/messages/invoke_transformations/test_anthropic_claude3_transformation.py b/tests/test_litellm/llms/bedrock/messages/invoke_transformations/test_anthropic_claude3_transformation.py index 5c5cae61a9..d3a9c94ea5 100644 --- a/tests/test_litellm/llms/bedrock/messages/invoke_transformations/test_anthropic_claude3_transformation.py +++ b/tests/test_litellm/llms/bedrock/messages/invoke_transformations/test_anthropic_claude3_transformation.py @@ -618,6 +618,147 @@ async def test_promote_message_stop_usage_preserves_message_delta_output_tokens( assert delta_out["usage"]["input_tokens"] == 3 +@pytest.mark.asyncio +async def test_promote_message_start_cache_when_message_stop_omits_cache_fields(): + """ + GovCloud / some Bedrock streams put cache_read only on message_start; delta and + stop repeat uncached input_tokens only. Merging start cache onto message_delta + avoids inconsistent usage and negative input costs (LIT-2411). + """ + cfg = AmazonAnthropicClaudeMessagesConfig() + + async def _stream(): # type: ignore[return-type] + yield { + "type": "message_start", + "message": { + "id": "msg_1", + "type": "message", + "role": "assistant", + "content": [], + "model": "claude-sonnet-4-5-20250929", + "usage": { + "input_tokens": 10, + "cache_creation_input_tokens": 0, + "cache_read_input_tokens": 22167, + "cache_creation": { + "ephemeral_5m_input_tokens": 0, + "ephemeral_1h_input_tokens": 0, + }, + "output_tokens": 4, + }, + }, + } + yield { + "type": "message_delta", + "delta": {"stop_reason": "end_turn", "stop_sequence": None}, + "usage": {"input_tokens": 10, "output_tokens": 181}, + } + yield {"type": "message_stop", "usage": {"input_tokens": 10, "output_tokens": 181}} + + merged: list[dict] = [] + async for chunk in cfg._promote_message_stop_usage(_stream()): + if isinstance(chunk, dict): + merged.append(chunk) + + delta_chunks = [c for c in merged if c.get("type") == "message_delta"] + assert len(delta_chunks) == 1 + u = delta_chunks[0]["usage"] + assert u["input_tokens"] == 10 + assert u["output_tokens"] == 181 + assert u["cache_read_input_tokens"] == 22167 + assert u["cache_creation_input_tokens"] == 0 + + +@pytest.mark.asyncio +async def test_unified_bedrock_messages_cache_on_start_only_never_negative_cost(): + """ + Regression guard for LIT-2411: + If cache usage is present only on message_start (and omitted from + message_delta/message_stop), final reconstructed usage + cost must still + be consistent and non-negative. + """ + from litellm import completion_cost + from litellm.proxy.pass_through_endpoints.llm_provider_handlers.anthropic_passthrough_logging_handler import ( + AnthropicPassthroughLoggingHandler, + ) + + cfg = AmazonAnthropicClaudeMessagesConfig() + + async def _stream(): # type: ignore[return-type] + yield { + "type": "message_start", + "message": { + "id": "msg_bdrk_01WuFzkDbE9KWgiWakMRNKcA", + "type": "message", + "role": "assistant", + "content": [], + "model": "claude-sonnet-4-5-20250929", + "stop_reason": None, + "stop_sequence": None, + "usage": { + "input_tokens": 10, + "cache_creation_input_tokens": 0, + "cache_read_input_tokens": 22167, + "cache_creation": { + "ephemeral_5m_input_tokens": 0, + "ephemeral_1h_input_tokens": 0, + }, + "output_tokens": 4, + }, + }, + } + yield {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}} + yield { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "text_delta", "text": "Hello from regression test"}, + } + yield {"type": "content_block_stop", "index": 0} + yield { + "type": "message_delta", + "delta": {"stop_reason": "end_turn", "stop_sequence": None}, + "usage": {"output_tokens": 181, "input_tokens": 10}, + } + yield {"type": "message_stop", "usage": {"input_tokens": 10, "output_tokens": 181}} + + logging_obj = LiteLLMLoggingObj( + model="bedrock/anthropic.claude-3-5-sonnet-20240620-v1:0", + messages=[{"role": "user", "content": "Hi"}], + stream=True, + call_type="chat", + start_time=datetime.now(), + litellm_call_id="test_cache_on_start_only_never_negative_cost", + function_id="test_cache_on_start_only_never_negative_cost", + ) + + collected: list[bytes] = [] + async for sse in cfg.bedrock_sse_wrapper( + completion_stream=_stream(), + litellm_logging_obj=logging_obj, + request_body={"model": "anthropic.claude-3-5-sonnet-20240620-v1:0"}, + ): + collected.append(sse) + + built = AnthropicPassthroughLoggingHandler._build_complete_streaming_response( + all_chunks=collected, + model="anthropic.claude-3-5-sonnet-20240620-v1:0", + litellm_logging_obj=Mock(), + ) + assert built.usage is not None + assert built.usage.prompt_tokens == 22177 + assert built.usage.completion_tokens == 181 + assert built.usage.cache_creation_input_tokens == 0 + assert built.usage.cache_read_input_tokens == 22167 + + cost = completion_cost( + completion_response=built, + model="bedrock/anthropic.claude-3-5-sonnet-20240620-v1:0", + custom_llm_provider="bedrock", + ) + assert cost > 0 + assert cost == pytest.approx(0.0093951, rel=0, abs=1e-9) + + @pytest.mark.asyncio async def test_unified_bedrock_messages_sse_usage_and_cost_claude_sonnet_46(): """