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

fix(github_copilot): surface copilot_usage token_details as cache token counts

Browse source
This commit is contained in:
Jeevan Babu Gotru 2026-10-27 10:20:41 -05:00 • committed by GitHub
parent 5bf0f7706f
commit 830314177f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 26 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
litellm/litellm_core_utils/streaming_handler.py
View file

@ -2757,10 +2757,36 @@ def calculate_total_usage(chunks: Sequence[ModelResponse | ModelResponseStream])
)
if latest_cost is not None:
returned_usage_chunk.cost = latest_cost
_apply_copilot_cache_counts(returned_usage_chunk, latest_usage_chunk)
return returned_usage_chunk
def _apply_copilot_cache_counts(returned_usage: Usage, latest_usage_chunk: Usage | Mapping[str, int]) -> None:
from litellm.llms.github_copilot.chat.transformation import copilot_cache_token_counts
copilot_usage: Final = (
latest_usage_chunk.get("copilot_usage")
if isinstance(latest_usage_chunk, Mapping)
else getattr(latest_usage_chunk, "copilot_usage", None)
)
if not isinstance(copilot_usage, Mapping):
return
cache_read, cache_write = copilot_cache_token_counts({"copilot_usage": copilot_usage})
if not cache_read and not cache_write:
return
details: Final = returned_usage.prompt_tokens_details or PromptTokensDetailsWrapper()
if cache_read and not getattr(details, "cached_tokens", None):
details.cached_tokens = cache_read
if cache_write and not getattr(details, "cache_write_tokens", None):
details.cache_write_tokens = cache_write
returned_usage.prompt_tokens_details = details # rebind-ok: attaches copilot cache counts to the aggregated usage
if cache_read and getattr(returned_usage, "cache_read_input_tokens", None) is None:
returned_usage.cache_read_input_tokens = cache_read
if cache_write and getattr(returned_usage, "cache_creation_input_tokens", None) is None:
returned_usage.cache_creation_input_tokens = cache_write
def generic_chunk_has_all_required_fields(chunk: dict) -> bool:
"""
Checks if the provided chunk dictionary contains all required fields for GenericStreamingChunk.