fix(anthropic): make opening usage estimates opt-in (#48633)

* fix(anthropic): make opening usage estimates opt-in

* docs(anthropic): follow README paragraph style

litellm/__init__.py

@@ -234,6 +234,7 @@ bedrock_neutralize_orphaned_tool_blocks: bool = True
 use_chat_completions_url_for_anthropic_messages: bool = bool(
     os.getenv("LITELLM_USE_CHAT_COMPLETIONS_URL_FOR_ANTHROPIC_MESSAGES", False)
 )  # When True, routes OpenAI /v1/messages requests to chat/completions instead of the Responses API
+anthropic_messages_estimate_initial_input_tokens: bool = False
 # When True, strip the OpenAI-flavored `usage.total_tokens` field that
 # LiteLLM injects into non-streaming /v1/messages responses, bringing the
 # wire response into line with the Anthropic spec (matches the streaming

litellm/anthropic_interface/readme.md

@@ -57,6 +57,40 @@ async for chunk in response:
     print(chunk)
 ```
 
+### Optional opening token estimate
+
+When `/v1/messages` streams through the Chat Completions adapter, the opening
+`message_start.usage.input_tokens` defaults to `0`. To show an estimated prompt
+size while a response streams, enable `anthropic_messages_estimate_initial_input_tokens`.
+This setting defaults to `false` and applies to all requests using this adapter
+
+For an existing proxy deployment, add the setting under `litellm_settings` in
+`config.yaml` and restart the proxy. Use a YAML boolean, without quotes
+
+```yaml
+litellm_settings:
+  anthropic_messages_estimate_initial_input_tokens: true
+```
+
+For the Python SDK, set the option before making streaming Messages requests
+
+```python
+import litellm
+
+litellm.anthropic_messages_estimate_initial_input_tokens = True
+```
+
+Send a streaming Messages request to a Chat Completions-backed model, such as a
+`hosted_vllm/` model. The opening `message_start` event includes the estimated
+input count. If estimation fails, the count remains `0`. Set the option to
+`false` in YAML or `False` in Python to disable estimation and skip token counting
+
+The opening count is approximate and is not suitable for billing or enforcing
+exact context limits. Opening output and cache counts remain `0`. Final
+`message_delta` usage still comes from the provider and takes precedence over the
+opening estimate. Native Anthropic streams, the Responses adapter, and
+non-streaming responses are unaffected. This setting does not select the adapter
+
 ### LiteLLM Proxy Server 
 
 
@@ -113,4 +147,4 @@ curl -L -X POST 'http://0.0.0.0:4000/v1/messages' \
   ],
   "max_tokens": 100
 }'
-```
+```

litellm/llms/anthropic/messages/adapters/streaming_iterator.py

@@ -761,13 +761,17 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
 
     def _create_initial_usage_delta(self) -> UsageDelta:
         """
-        Estimate prompt tokens for message_start without consuming upstream chunks.
+        Optionally estimate prompt tokens without consuming upstream chunks.
 
         Output and cache counts start at zero. Provider usage in message_delta
         remains authoritative. Failed estimates fall back to zero.
         """
         return UsageDelta(
-            input_tokens=self._estimate_initial_input_tokens(),
+            input_tokens=(
+                self._estimate_initial_input_tokens()
+                if litellm.anthropic_messages_estimate_initial_input_tokens is True
+                else 0
+            ),
             output_tokens=0,
             cache_creation_input_tokens=0,
             cache_read_input_tokens=0,

tests/unit/llms/anthropic/messages/adapters/test_streaming_iterator_message_id.py

@@ -9,6 +9,7 @@ import datetime
 import json
 from collections.abc import AsyncIterator, Iterator, Mapping, Sequence
 from typing import Final
+from unittest.mock import Mock
 
 import pytest
 import respx
@@ -24,7 +25,6 @@ from litellm.llms.anthropic.messages.adapters.streaming_iterator import (
 from litellm.types.llms.openai import AllMessageValues, ChatCompletionToolParam
 from litellm.types.utils import ModelResponseStream
 
-
 _EVENT: Final = TypeAdapter(Mapping[str, object])
 _MESSAGES: Final = TypeAdapter(list[AllMessageValues])
 _TOOLS: Final = TypeAdapter(list[ChatCompletionToolParam])
@@ -42,7 +42,9 @@ _BLOCK_PROMPTS: Final[tuple[list[AllMessageValues], ...]] = (
 )
 
 
-def _assert_estimated_start_and_provider_final_usage(raw_events: Sequence[bytes], request_body: bytes) -> None:
+def _assert_start_and_provider_final_usage(
+    raw_events: Sequence[bytes], request_body: bytes, estimate_enabled: bool
+) -> None:
     request: Final = _EVENT.validate_json(request_body)
     events: Final = tuple(_EVENT.validate_json(chunk.decode().split("data: ", 1)[1]) for chunk in raw_events)
     start: Final = _EVENT.validate_python(
@@ -56,7 +58,7 @@ def _assert_estimated_start_and_provider_final_usage(raw_events: Sequence[bytes]
     )
     assert expected > 0
     assert start["usage"] == {
-        "input_tokens": expected,
+        "input_tokens": expected if estimate_enabled else 0,
         "output_tokens": 0,
         "cache_creation_input_tokens": 0,
         "cache_read_input_tokens": 0,
@@ -64,7 +66,12 @@ def _assert_estimated_start_and_provider_final_usage(raw_events: Sequence[bytes]
     assert final["usage"] == {"input_tokens": 3, "output_tokens": 4}
 
 
-def test_sync_message_start_estimates_prompt_without_replacing_provider_usage(respx_mock: respx.MockRouter) -> None:
+@pytest.mark.parametrize("estimate_enabled", [None, False, True, "false"])
+def test_sync_message_start_opt_in_preserves_provider_final_usage(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, estimate_enabled: bool | str | None
+) -> None:
+    if estimate_enabled is not None:
+        monkeypatch.setattr(litellm, "anthropic_messages_estimate_initial_input_tokens", estimate_enabled)
     stream: Final = LiteLLMMessagesToCompletionTransformationHandler.anthropic_messages_handler(
         max_tokens=16,
         messages=MESSAGES,
@@ -74,13 +81,18 @@ def test_sync_message_start_estimates_prompt_without_replacing_provider_usage(re
         stream=True,
         custom_llm_provider="groq",
     )
-    _assert_estimated_start_and_provider_final_usage(list(stream), respx_mock.calls.last.request.content)
+    _assert_start_and_provider_final_usage(
+        list(stream), respx_mock.calls.last.request.content, estimate_enabled is True
+    )
 
 
 @pytest.mark.asyncio
-async def test_async_message_start_estimates_prompt_without_replacing_provider_usage(
-    respx_mock: respx.MockRouter,
+@pytest.mark.parametrize("estimate_enabled", [None, False, True, "false"])
+async def test_async_message_start_opt_in_preserves_provider_final_usage(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, estimate_enabled: bool | str | None
 ) -> None:
+    if estimate_enabled is not None:
+        monkeypatch.setattr(litellm, "anthropic_messages_estimate_initial_input_tokens", estimate_enabled)
     stream: Final = await LiteLLMMessagesToCompletionTransformationHandler.async_anthropic_messages_handler(
         max_tokens=16,
         messages=MESSAGES,
@@ -91,7 +103,7 @@ async def test_async_message_start_estimates_prompt_without_replacing_provider_u
         custom_llm_provider="groq",
     )
     raw: Final = [chunk async for chunk in stream]
-    _assert_estimated_start_and_provider_final_usage(raw, respx_mock.calls.last.request.content)
+    _assert_start_and_provider_final_usage(raw, respx_mock.calls.last.request.content, estimate_enabled is True)
 
 
 class _FailingUpstream:
@@ -102,10 +114,31 @@ class _FailingUpstream:
         raise RuntimeError("upstream failed before its first chunk")
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("is_async", [False, True])
+async def test_disabled_initial_estimate_skips_token_counting(monkeypatch: pytest.MonkeyPatch, is_async: bool) -> None:
+    counter: Final = Mock(side_effect=AssertionError("disabled estimation must not count tokens"))
+    monkeypatch.setattr(litellm, "anthropic_messages_estimate_initial_input_tokens", False)
+    monkeypatch.setattr(litellm, "token_counter", counter)
+    wrapper: Final = AnthropicStreamWrapper(
+        _FailingUpstream(), model="kimi-k2", completion_request={"messages": MESSAGES}
+    )
+    first: Final = _EVENT.validate_python(await anext(wrapper) if is_async else next(wrapper))
+    message: Final = _EVENT.validate_python(first["message"])
+    assert message["usage"] == {
+        "input_tokens": 0,
+        "output_tokens": 0,
+        "cache_creation_input_tokens": 0,
+        "cache_read_input_tokens": 0,
+    }
+    counter.assert_not_called()
+
+
 @pytest.mark.parametrize("messages", _BLOCK_PROMPTS)
 def test_sync_initial_estimate_counts_content_blocks(
-    messages: list[AllMessageValues], respx_mock: respx.MockRouter
+    messages: list[AllMessageValues], respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
 ) -> None:
+    monkeypatch.setattr(litellm, "anthropic_messages_estimate_initial_input_tokens", True)
     image: Final = respx_mock.get("https://image.example/test.png").respond(503)
     wrapper: Final = AnthropicStreamWrapper(iter([]), model="gpt-4o", completion_request={"messages": messages})
     expected: Final = litellm.token_counter(model="gpt-4o", messages=messages, use_default_image_token_count=True)
@@ -123,8 +156,9 @@ def test_sync_initial_estimate_counts_content_blocks(
 @pytest.mark.asyncio
 @pytest.mark.parametrize("messages", _BLOCK_PROMPTS)
 async def test_async_initial_estimate_counts_content_blocks(
-    messages: list[AllMessageValues], respx_mock: respx.MockRouter
+    messages: list[AllMessageValues], respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
 ) -> None:
+    monkeypatch.setattr(litellm, "anthropic_messages_estimate_initial_input_tokens", True)
     image: Final = respx_mock.get("https://image.example/test.png").respond(503)
     wrapper: Final = AnthropicStreamWrapper(iter([]), model="gpt-4o", completion_request={"messages": messages})
     expected: Final = litellm.token_counter(model="gpt-4o", messages=messages, use_default_image_token_count=True)
@@ -140,7 +174,10 @@ async def test_async_initial_estimate_counts_content_blocks(
 
 
 @pytest.mark.parametrize("prompt", [{}, {"messages": []}, {"messages": "invalid"}])
-def test_sync_initial_estimate_fallback_does_not_consume_upstream(prompt: Mapping[str, object]) -> None:
+def test_sync_initial_estimate_fallback_does_not_consume_upstream(
+    prompt: Mapping[str, object], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "anthropic_messages_estimate_initial_input_tokens", True)
     wrapper: Final = AnthropicStreamWrapper(_FailingUpstream(), model="kimi-k2", completion_request=prompt)
     first: Final = _EVENT.validate_python(next(wrapper))
     message: Final = _EVENT.validate_python(first["message"])
@@ -155,7 +192,10 @@ def test_sync_initial_estimate_fallback_does_not_consume_upstream(prompt: Mappin
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("prompt", [{}, {"messages": []}, {"messages": "invalid"}])
-async def test_async_initial_estimate_fallback_does_not_consume_upstream(prompt: Mapping[str, object]) -> None:
+async def test_async_initial_estimate_fallback_does_not_consume_upstream(
+    prompt: Mapping[str, object], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "anthropic_messages_estimate_initial_input_tokens", True)
     wrapper: Final = AnthropicStreamWrapper(_FailingUpstream(), model="kimi-k2", completion_request=prompt)
     first: Final = _EVENT.validate_python(await anext(wrapper))
     message: Final = _EVENT.validate_python(first["message"])