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

fix(anthropic): populate inference_geo from request for cost tracking

Browse source 
Anthropic does not echo inference_geo back on response usage, so regional pricing multipliers almost never applied. Fall back to the requested region (optional_params / litellm_params / pass-through request body) when the provider omits it, and keep it through streaming reassembly.

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
shivam 2026-07-28 00:06:06 +00:00 • committed by Devin AI
parent 1d3b64c66f
commit 5b59e6b80c
9 changed files with 359 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
litellm/litellm_core_utils/litellm_logging.py
View file

@ -3431,6 +3431,9 @@ class Logging(LiteLLMLoggingBaseClass):
if isinstance(result, ResponsesAPIResponse):
return self._translate_responses_api_response_to_model_response(result)
request_params: Final = self.model_call_details.get("optional_params") or {}
deployment_params: Final = self.model_call_details.get("litellm_params") or {}
httpx_response: Final = self.model_call_details.get("httpx_response", None)
if httpx_response and isinstance(httpx_response, httpx.Response):
result = litellm.AnthropicConfig().transform_response(
@ -3439,12 +3442,12 @@ class Logging(LiteLLMLoggingBaseClass):
model=self.model,
messages=[],
logging_obj=self,
optional_params={},
optional_params=request_params,
api_key="",
request_data={},
encoding=litellm.encoding,
json_mode=False,
litellm_params={},
litellm_params=deployment_params,
)
else:
from litellm.types.llms.anthropic import AnthropicResponse
@ -3460,6 +3463,9 @@ class Logging(LiteLLMLoggingBaseClass):
),
model_response=litellm.ModelResponse(),
json_mode=None,
speed=request_params.get("speed"),
inference_geo=litellm.AnthropicConfig.get_request_inference_geo(request_params)
or litellm.AnthropicConfig.get_request_inference_geo(deployment_params),
)
return result

13
litellm/litellm_core_utils/streaming_chunk_builder_utils.py
View file

@ -730,6 +730,7 @@ class ChunkProcessor:
# lost and 1h cache writes get billed at the 5m rate.
cache_creation_token_details: CacheCreationTokenDetails | None = None
cost: float | None = None
inference_geo: str | None = None
for chunk in chunks:
usage_chunk = self._extract_usage_chunk(chunk)
@ -785,6 +786,10 @@ class ChunkProcessor:
if usage_chunk_dict["cost"] is not None:
cost = usage_chunk_dict["cost"]
chunk_inference_geo = getattr(usage_chunk, "inference_geo", None)
if isinstance(chunk_inference_geo, str) and chunk_inference_geo:
inference_geo = chunk_inference_geo
prompt_tokens_details = attach_cache_creation_token_details(prompt_tokens_details, cache_creation_token_details)
completion_tokens = self._reset_anthropic_cursor_completion_tokens(
@ -803,6 +808,7 @@ class ChunkProcessor:
completion_tokens_details=completion_tokens_details,
prompt_tokens_details=prompt_tokens_details,
cost=cost,
inference_geo=inference_geo,
)
@staticmethod
@ -872,6 +878,7 @@ class ChunkProcessor:
]
prompt_tokens_details: PromptTokensDetailsWrapper | None = calculated_usage_per_chunk["prompt_tokens_details"]
cost: Final[float | None] = calculated_usage_per_chunk["cost"]
inference_geo: Final[str | None] = calculated_usage_per_chunk["inference_geo"]
try:
returned_usage.prompt_tokens = prompt_tokens or token_counter(model=model, messages=messages)
@ -932,9 +939,13 @@ class ChunkProcessor:
if cost is not None:
setattr(returned_usage, "cost", cost)
extra_usage_fields: Final[dict[str, str]] = (
{"inference_geo": inference_geo} if inference_geo is not None else {}
)
# Return a new usage object with the new values
returned_usage = Usage(**returned_usage.model_dump())
returned_usage = Usage(**returned_usage.model_dump(), **extra_usage_fields)
return returned_usage

13
litellm/llms/anthropic/chat/handler.py
View file

@ -73,6 +73,7 @@ async def make_call(
timeout: float | httpx.Timeout | None,
json_mode: bool,
speed: str | None = None,
inference_geo: str | None = None
tool_name_reverse_map: dict[str, str] | None = None,
) -> tuple[Any, httpx.Headers]:
if client is None:
@ -108,6 +109,7 @@ async def make_call(
sync_stream=False,
json_mode=json_mode,
speed=speed,
inference_geo=inference_geo,
tool_name_reverse_map=tool_name_reverse_map,
)
@ -133,6 +135,7 @@ def make_sync_call(
timeout: float | httpx.Timeout | None,
json_mode: bool,
speed: str | None = None,
inference_geo: str | None = None
tool_name_reverse_map: dict[str, str] | None = None,
) -> tuple[Any, httpx.Headers]:
if client is None:
@ -176,6 +179,7 @@ def make_sync_call(
sync_stream=True,
json_mode=json_mode,
speed=speed,
inference_geo=inference_geo,
tool_name_reverse_map=tool_name_reverse_map,
)
@ -231,6 +235,8 @@ class AnthropicChatCompletion(BaseLLM):
timeout=timeout,
json_mode=json_mode,
speed=optional_params.get("speed") if optional_params else None,
inference_geo=AnthropicConfig.get_request_inference_geo(optional_params)
or AnthropicConfig.get_request_inference_geo(litellm_params if isinstance(litellm_params, dict) else None),
tool_name_reverse_map=(
litellm_params.get(ANTHROPIC_TOOL_NAME_REVERSE_MAP_KEY) if isinstance(litellm_params, dict) else None
),
@ -452,6 +458,10 @@ class AnthropicChatCompletion(BaseLLM):
timeout=timeout,
json_mode=json_mode,
speed=optional_params.get("speed") if optional_params else None,
inference_geo=AnthropicConfig.get_request_inference_geo(optional_params)
or AnthropicConfig.get_request_inference_geo(
litellm_params if isinstance(litellm_params, dict) else None
),
tool_name_reverse_map=(
litellm_params.get(ANTHROPIC_TOOL_NAME_REVERSE_MAP_KEY)
if isinstance(litellm_params, dict)
@ -521,6 +531,7 @@ class ModelResponseIterator:
sync_stream: bool,
json_mode: bool | None = False,
speed: str | None = None,
inference_geo: str | None = None
tool_name_reverse_map: dict[str, str] | None = None,
):
self.streaming_response = streaming_response
@ -529,6 +540,7 @@ class ModelResponseIterator:
self.tool_index = -1
self.json_mode = json_mode
self.speed = speed
self.inference_geo = inference_geo
# rewritten-name -> caller's original. Built per-request from the
# forward map in AnthropicConfig._build_request_tool_name_maps; only
# contains entries we actually rewrote, so a tool legitimately named
@ -599,6 +611,7 @@ class ModelResponseIterator:
usage_object=cast(dict, anthropic_usage_chunk),
reasoning_content=reasoning_content,
speed=self.speed,
inference_geo=self.inference_geo,
)
def _content_block_delta_helper(

28
litellm/llms/anthropic/chat/transformation.py
View file

@ -1,6 +1,7 @@
import json
import re
import time
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final, NoReturn, cast
import httpx
@ -2117,12 +2118,27 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
return False
return any(key in usage_object for key in ("cache_read_input_tokens", "cache_creation_input_tokens"))
@staticmethod
def get_request_inference_geo(params: Mapping[str, Any] | None) -> str | None:
"""
Read the region a request asked to be served from.
Anthropic echoes ``inference_geo`` on response usage for some surfaces but not all, and the
streamed usage LiteLLM reassembles can lose it, so the requested region keeps geo pricing
keyed on something when the response does not carry one.
"""
if not params:
return None
value: Final = params.get("inference_geo")
return value if isinstance(value, str) and value else None
def calculate_usage(
self,
usage_object: dict,
reasoning_content: str | None,
completion_response: dict | None = None,
speed: str | None = None,
inference_geo: str | None = None,
) -> Usage:
# NOTE: Sometimes the usage object has None set explicitly for token counts, meaning .get() & key access returns None, and we need to account for this
raw_prompt_tokens: Final = usage_object.get("input_tokens", 0) or 0
@ -2135,9 +2151,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
cache_creation_token_details: CacheCreationTokenDetails | None = None
web_search_requests: int | None = None
tool_search_requests: int | None = None
inference_geo: str | None = None
if "inference_geo" in _usage and _usage["inference_geo"] is not None:
inference_geo = _usage["inference_geo"]
resolved_inference_geo: Final = AnthropicConfig.get_request_inference_geo(_usage) or inference_geo
service_tier: Final = cast(
str | None,
_usage.get("service_tier"),
@ -2223,7 +2237,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
if (web_search_requests is not None or tool_search_requests is not None)
else None
),
inference_geo=inference_geo,
inference_geo=resolved_inference_geo,
speed=speed,
service_tier=service_tier,
)
@ -2317,6 +2331,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
json_mode: bool | None = None,
prefix_prompt: str | None = None,
speed: str | None = None,
inference_geo: str | None = None,
tool_name_reverse_map: dict[str, str] | None = None,
):
_hidden_params: Final[dict] = {}
@ -2405,6 +2420,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
reasoning_content=reasoning_content,
completion_response=completion_response,
speed=speed,
inference_geo=inference_geo,
)
setattr(model_response, "usage", usage)
@ -2469,6 +2485,9 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
prefix_prompt: Final = self.get_prefix_prompt(messages=messages)
speed: Final = optional_params.get("speed")
inference_geo: Final = self.get_request_inference_geo(optional_params) or self.get_request_inference_geo(
litellm_params
)
tool_name_reverse_map: dict[str, str] | None = None
if isinstance(litellm_params, dict):
_candidate: Final = litellm_params.get(ANTHROPIC_TOOL_NAME_REVERSE_MAP_KEY)
@ -2482,6 +2501,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
json_mode=json_mode,
prefix_prompt=prefix_prompt,
speed=speed,
inference_geo=inference_geo,
tool_name_reverse_map=tool_name_reverse_map,
)
return model_response

20
litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py
View file

@ -74,6 +74,7 @@ class AnthropicPassthroughLoggingHandler:
)
model: Final = response_body.get("model", "")
resolved_request_body: Final = request_body or kwargs.get("request_body") or {}
anthropic_config: Final = get_anthropic_config(url_route)
litellm_model_response: Final[ModelResponse] = anthropic_config().transform_response(
raw_response=httpx_response,
@ -81,7 +82,7 @@ class AnthropicPassthroughLoggingHandler:
model=model,
messages=[],
logging_obj=logging_obj,
optional_params={},
optional_params=resolved_request_body,
api_key="",
request_data={},
encoding=litellm.encoding,
@ -335,6 +336,7 @@ class AnthropicPassthroughLoggingHandler:
all_chunks=all_chunks,
litellm_logging_obj=litellm_logging_obj,
model=model,
request_inference_geo=AnthropicConfig.get_request_inference_geo(request_body),
)
except Exception as e:
# stream_chunk_builder re-raises assembly failures (as litellm.APIError)
@ -356,6 +358,7 @@ class AnthropicPassthroughLoggingHandler:
complete_streaming_response = AnthropicPassthroughLoggingHandler._build_usage_only_response_from_chunks(
all_chunks=all_chunks,
model=model,
request_inference_geo=AnthropicConfig.get_request_inference_geo(request_body),
)
except Exception as e:
verbose_proxy_logger.warning(
@ -420,6 +423,7 @@ class AnthropicPassthroughLoggingHandler:
all_chunks: Sequence[str | bytes],
litellm_logging_obj: LiteLLMLoggingObj,
model: str,
request_inference_geo: str | None = None,
) -> ModelResponse | TextCompletionResponse | None:
"""
Builds complete response from raw Anthropic chunks.
@ -439,16 +443,11 @@ class AnthropicPassthroughLoggingHandler:
text run removes O(num_output_tokens) of it.
"""
collapsed: Final = AnthropicPassthroughLoggingHandler._collapse_pure_text_chunks(all_chunks)
if collapsed is not None:
return AnthropicPassthroughLoggingHandler._build_complete_streaming_response_legacy(
all_chunks=collapsed,
litellm_logging_obj=litellm_logging_obj,
model=model,
)
return AnthropicPassthroughLoggingHandler._build_complete_streaming_response_legacy(
all_chunks=all_chunks,
all_chunks=collapsed if collapsed is not None else all_chunks,
litellm_logging_obj=litellm_logging_obj,
model=model,
request_inference_geo=request_inference_geo,
)
# Anthropic SSE block/delta types that the fast path is NOT allowed to
@ -576,6 +575,7 @@ class AnthropicPassthroughLoggingHandler:
all_chunks: Sequence[str | bytes],
litellm_logging_obj: LiteLLMLoggingObj,
model: str,
request_inference_geo: str | None = None,
) -> ModelResponse | TextCompletionResponse | None:
"""
Original reconstruction: convert every SSE event to a generic chunk
@ -591,6 +591,7 @@ class AnthropicPassthroughLoggingHandler:
anthropic_model_response_iterator: Final = AnthropicModelResponseIterator(
streaming_response=None,
sync_stream=False,
inference_geo=request_inference_geo,
)
all_openai_chunks: Final = []
@ -650,6 +651,7 @@ class AnthropicPassthroughLoggingHandler:
def _build_usage_only_response_from_chunks(
all_chunks: Sequence[str | bytes],
model: str,
request_inference_geo: str | None = None,
) -> ModelResponse | None:
"""
Build a usage-bearing ModelResponse from Anthropic SSE token-usage events, for
@ -668,7 +670,7 @@ class AnthropicPassthroughLoggingHandler:
output_tokens = 0
web_search_requests: int | None = None
tool_search_requests: int | None = None
inference_geo: str | None = None
inference_geo: str | None = request_inference_geo
stop_reason: str | None = None
found_usage = False
resolved_model = model

1
litellm/types/litellm_core_utils/streaming_chunk_builder_utils.py
View file

@ -13,3 +13,4 @@ class UsagePerChunk(TypedDict):
completion_tokens_details: CompletionTokensDetails | None
prompt_tokens_details: PromptTokensDetailsWrapper | None
cost: float | None
inference_geo: str | None

36
tests/test_litellm/litellm_core_utils/test_streaming_chunk_builder_utils.py
View file

@ -1223,3 +1223,39 @@ def test_get_combined_tool_content_joins_many_custom_tool_input_fragments_in_ord
assert isinstance(combined[1], ChatCompletionMessageCustomToolCall)
assert combined[1].custom.name == "run_script"
assert combined[1].custom.input == "".join(object_fragments)
def test_inference_geo_preserved_across_streamed_usage_chunks():
"""
Regression: reassembled streaming usage dropped inference_geo, so regional
pricing never applied to streamed Anthropic requests.
"""
chunk1 = ModelResponseStream(
id="chatcmpl-1",
created=1745513206,
model="claude-opus-4-8",
choices=[
StreamingChoices(finish_reason=None, index=0, delta=Delta(content="Hi"))
],
usage=Usage(
completion_tokens=1, prompt_tokens=10, total_tokens=11, inference_geo="us"
),
)
chunk2 = ModelResponseStream(
id="chatcmpl-1",
created=1745513207,
model="claude-opus-4-8",
choices=[
StreamingChoices(finish_reason="stop", index=0, delta=Delta(content=""))
],
usage=Usage(
completion_tokens=5, prompt_tokens=10, total_tokens=15, inference_geo="us"
),
)
usage = ChunkProcessor(chunks=[chunk1, chunk2]).calculate_usage(
chunks=[chunk1, chunk2], model="claude-opus-4-8", completion_output="Hi"
)
assert usage.inference_geo == "us"

102
tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_transformation.py
View file

@ -5883,3 +5883,105 @@ def test_is_anthropic_usage_object_rejects_responses_api_usage():
"output_tokens_details": {"reasoning_tokens": 0},
}
)
ANTHROPIC_GEO_RESPONSE = {
"id": "msg_geo",
"type": "message",
"role": "assistant",
"model": "claude-opus-4-8",
"content": [{"type": "text", "text": "hi"}],
"stop_reason": "end_turn",
"usage": {"input_tokens": 1000, "output_tokens": 1000},
}
def test_calculate_usage_falls_back_to_requested_inference_geo():
"""
Anthropic does not echo ``inference_geo`` back on response usage, so the
region requested by the caller must be used for geo pricing.
"""
usage = AnthropicConfig().calculate_usage(
usage_object={"input_tokens": 10, "output_tokens": 5},
reasoning_content=None,
inference_geo="us",
)
assert usage.inference_geo == "us"
def test_calculate_usage_prefers_response_inference_geo_over_request():
usage = AnthropicConfig().calculate_usage(
usage_object={"input_tokens": 10, "output_tokens": 5, "inference_geo": "eu"},
reasoning_content=None,
inference_geo="us",
)
assert usage.inference_geo == "eu"
def test_transform_response_populates_inference_geo_from_request():
"""Regression: /chat/completions requests with inference_geo were billed without the geo uplift."""
import httpx
raw_response = httpx.Response(status_code=200, json=ANTHROPIC_GEO_RESPONSE)
model_response = AnthropicConfig().transform_response(
model="claude-opus-4-8",
raw_response=raw_response,
model_response=litellm.ModelResponse(),
logging_obj=MagicMock(),
request_data={},
messages=[{"role": "user", "content": "hi"}],
optional_params={"inference_geo": "us"},
litellm_params={},
encoding=None,
)
assert model_response.usage.inference_geo == "us"
model_info = litellm.get_model_info("claude-opus-4-8", custom_llm_provider="anthropic")
base_cost = 1000 * model_info["input_cost_per_token"] + 1000 * model_info["output_cost_per_token"]
geo_multiplier = model_info["provider_specific_entry"]["us"]
geo_cost = litellm.completion_cost(completion_response=model_response, model="claude-opus-4-8")
assert geo_cost == pytest.approx(base_cost * geo_multiplier)
def test_transform_response_reads_inference_geo_from_litellm_params():
"""Deployment-level `inference_geo` (litellm_params) must also reach the usage object."""
import httpx
raw_response = httpx.Response(status_code=200, json=ANTHROPIC_GEO_RESPONSE)
model_response = AnthropicConfig().transform_response(
model="claude-opus-4-8",
raw_response=raw_response,
model_response=litellm.ModelResponse(),
logging_obj=MagicMock(),
request_data={},
messages=[{"role": "user", "content": "hi"}],
optional_params={},
litellm_params={"inference_geo": "us"},
encoding=None,
)
assert model_response.usage.inference_geo == "us"
def test_streaming_iterator_populates_inference_geo_from_request():
"""Streamed usage chunks must carry the requested region for geo pricing."""
from litellm.llms.anthropic.chat.handler import ModelResponseIterator
iterator = ModelResponseIterator(streaming_response=None, sync_stream=True, inference_geo="us")
chunk = iterator.chunk_parser(
{
"type": "message_delta",
"delta": {"stop_reason": "end_turn"},
"usage": {"output_tokens": 20},
}
)
assert chunk.usage.inference_geo == "us"

152
tests/test_litellm/proxy/pass_through_endpoints/llm_provider_handlers/test_anthropic_passthrough_logging_handler.py
View file

@ -2192,3 +2192,155 @@ class TestAnthropicResponseCostRecordedOnModelCallDetails:
logging_obj.model_call_details["response_cost"] == kwargs["response_cost"]
)
assert logging_obj.model_call_details["response_cost"] > 0
class TestPassthroughInferenceGeo:
"""
Anthropic never echoes `inference_geo` back on response usage, so pass-through
requests routed to a paid region were priced without their geo multiplier.
The requested region must be carried onto the usage object instead.
"""
_MODEL = "claude-opus-4-8"
@staticmethod
def _sse(data: dict) -> bytes:
return f"event: {data['type']}\ndata: {json.dumps(data)}\n\n".encode()
@classmethod
def _chunks(cls) -> list:
frames = [
cls._sse(
{
"type": "message_start",
"message": {
"id": "msg_geo",
"type": "message",
"role": "assistant",
"model": cls._MODEL,
"content": [],
"stop_reason": None,
"usage": {"input_tokens": 1000, "output_tokens": 1},
},
}
),
cls._sse(
{
"type": "content_block_start",
"index": 0,
"content_block": {"type": "text", "text": ""},
}
),
cls._sse(
{
"type": "content_block_delta",
"index": 0,
"delta": {"type": "text_delta", "text": "Hello"},
}
),
cls._sse({"type": "content_block_stop", "index": 0}),
cls._sse(
{
"type": "message_delta",
"delta": {"stop_reason": "end_turn"},
"usage": {"output_tokens": 1000},
}
),
cls._sse({"type": "message_stop"}),
]
from litellm.proxy.pass_through_endpoints.streaming_handler import (
PassThroughStreamingHandler,
)
return PassThroughStreamingHandler._convert_raw_bytes_to_str_lines(frames)
def _run_streaming_handler(self, request_body: dict):
from litellm.types.passthrough_endpoints.pass_through_endpoints import (
EndpointType,
)
logging_obj = LiteLLMLoggingObj(
model=self._MODEL,
messages=[{"role": "user", "content": "hello"}],
stream=True,
call_type="pass_through_endpoint",
start_time=datetime.now(),
litellm_call_id="test-call-id",
function_id="1245",
)
logging_obj.model_call_details["stream"] = True
return AnthropicPassthroughLoggingHandler._handle_logging_anthropic_collected_chunks(
litellm_logging_obj=logging_obj,
passthrough_success_handler_obj=MagicMock(),
url_route="/anthropic/v1/messages",
request_body=request_body,
endpoint_type=EndpointType.ANTHROPIC,
start_time=datetime.now(),
all_chunks=self._chunks(),
end_time=datetime.now(),
)
def test_streaming_usage_carries_requested_inference_geo(self):
result = self._run_streaming_handler(
{"model": self._MODEL, "stream": True, "inference_geo": "us"}
)
assert result["result"].usage.inference_geo == "us"
def test_streaming_usage_without_requested_geo_stays_unset(self):
result = self._run_streaming_handler({"model": self._MODEL, "stream": True})
assert result["result"].usage.get("inference_geo") is None
def test_streaming_response_cost_includes_geo_multiplier(self):
geo_cost = self._run_streaming_handler(
{"model": self._MODEL, "stream": True, "inference_geo": "us"}
)["kwargs"]["response_cost"]
base_cost = self._run_streaming_handler(
{"model": self._MODEL, "stream": True}
)["kwargs"]["response_cost"]
assert geo_cost == pytest.approx(base_cost * 1.1)
def test_usage_only_fallback_carries_requested_inference_geo(self):
response = (
AnthropicPassthroughLoggingHandler._build_usage_only_response_from_chunks(
all_chunks=self._chunks(),
model=self._MODEL,
request_inference_geo="us",
)
)
assert response is not None
assert response.usage.inference_geo == "us"
def test_provider_reported_geo_wins_over_requested_geo(self):
chunks = [
self._sse(
{
"type": "message_start",
"message": {
"id": "msg_geo",
"type": "message",
"role": "assistant",
"model": self._MODEL,
"content": [],
"usage": {
"input_tokens": 10,
"output_tokens": 1,
"inference_geo": "eu",
},
},
}
),
]
response = (
AnthropicPassthroughLoggingHandler._build_usage_only_response_from_chunks(
all_chunks=chunks, model=self._MODEL, request_inference_geo="us"
)
)
assert response is not None
assert response.usage.inference_geo == "eu"