scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-13 23:11:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 101

style: satisfy UP045 and LIT002 gates in anthropic thinking token path

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
milan 2026-07-31 04:41:54 +00:00
parent cd54183e62
commit da502df67a
2 changed files with 10 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -140,7 +140,7 @@ _ANTHROPIC_TOOL_NAME_MAX_LEN = 128
ANTHROPIC_TOOL_NAME_REVERSE_MAP_KEY = "_anthropic_tool_name_map"
def _reported_thinking_tokens(usage_object: Mapping[str, Any]) -> Optional[int]:
def _reported_thinking_tokens(usage_object: Mapping[str, Any]) -> int | None:
"""Read the provider-reported reasoning token count from an Anthropic usage object.
Anthropic (and Bedrock's Anthropic-compatible surface) reports extended-thinking
@ -156,7 +156,7 @@ def _reported_thinking_tokens(usage_object: Mapping[str, Any]) -> Optional[int]:
return thinking_tokens
def _sum_reported_thinking_tokens(iterations: Sequence[Any]) -> Optional[int]:
def _sum_reported_thinking_tokens(iterations: Sequence[Any]) -> int | None:
reported = tuple(_reported_thinking_tokens(iteration) for iteration in iterations if isinstance(iteration, Mapping))
present = tuple(tokens for tokens in reported if tokens is not None)
return sum(present) if present else None

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

@ -1,4 +1,5 @@
import json
from collections.abc import Mapping
from datetime import datetime
from typing import TYPE_CHECKING, Any, List, Optional, Sequence, Union, cast
@ -669,7 +670,7 @@ class AnthropicPassthroughLoggingHandler:
tool_search_requests: Optional[int] = None
inference_geo: Optional[str] = None
stop_reason: Optional[str] = None
thinking_tokens: Optional[int] = None
output_tokens_details: Mapping[str, Any] | None = None
found_usage = False
resolved_model = model
for _chunk_str in all_chunks:
@ -695,8 +696,8 @@ class AnthropicPassthroughLoggingHandler:
if usage.get("output_tokens") is not None:
output_tokens = usage.get("output_tokens")
_otd = usage.get("output_tokens_details")
if isinstance(_otd, dict) and _otd.get("thinking_tokens") is not None:
thinking_tokens = _otd.get("thinking_tokens")
if isinstance(_otd, Mapping) and _otd.get("thinking_tokens") is not None:
output_tokens_details = _otd
found_usage = True
elif event_type == "message_delta":
_delta_stop = (data.get("delta") or {}).get("stop_reason")
@ -716,8 +717,8 @@ class AnthropicPassthroughLoggingHandler:
if usage.get("inference_geo") is not None:
inference_geo = usage.get("inference_geo")
_otd = usage.get("output_tokens_details")
if isinstance(_otd, dict) and _otd.get("thinking_tokens") is not None:
thinking_tokens = _otd.get("thinking_tokens")
if isinstance(_otd, Mapping) and _otd.get("thinking_tokens") is not None:
output_tokens_details = _otd
found_usage = True
if not found_usage:
return None
@ -749,8 +750,8 @@ class AnthropicPassthroughLoggingHandler:
usage_object["server_tool_use"] = _server_tool_use
if inference_geo is not None:
usage_object["inference_geo"] = inference_geo
if thinking_tokens is not None:
usage_object["output_tokens_details"] = {"thinking_tokens": thinking_tokens}
if output_tokens_details is not None:
usage_object["output_tokens_details"] = output_tokens_details
usage_obj = AnthropicConfig().calculate_usage(usage_object=usage_object, reasoning_content=None)
return ModelResponse(
model=resolved_model,