fix(gigachat): improve usage reporting and config handling

litellm/llms/gigachat/authenticator.py

@@ -65,6 +65,7 @@ def get_access_token(
     credentials: Optional[str] = None,
     scope: Optional[str] = None,
     auth_url: Optional[str] = None,
+    litellm_params: Optional[dict] = None,
 ) -> str:
     """
     Get valid access token, using cache if available.
@@ -80,6 +81,15 @@ def get_access_token(
     Raises:
         GigaChatAuthError: If authentication fails
     """
+    if not litellm_params:
+        litellm_params = {}
+
+    access_token = litellm_params.get("gigachat_access_token") or get_secret_str(
+        "GIGACHAT_ACCESS_TOKEN"
+    )
+    if access_token:
+        return access_token
+
     credentials = credentials or _get_credentials()
     if not credentials:
         raise GigaChatAuthError(
@@ -87,8 +97,8 @@ def get_access_token(
             message="GigaChat credentials not provided. Set GIGACHAT_CREDENTIALS or GIGACHAT_API_KEY environment variable.",
         )
 
-    scope = scope or _get_scope()
-    auth_url = auth_url or _get_auth_url()
+    scope = scope or litellm_params.get("gigachat_scope") or _get_scope()
+    auth_url = auth_url or litellm_params.get("gigachat_auth_url") or _get_auth_url()
 
     # Check cache
     cache_key = f"gigachat_token:{credentials[:16]}"
@@ -117,6 +127,7 @@ async def get_access_token_async(
     credentials: Optional[str] = None,
     scope: Optional[str] = None,
     auth_url: Optional[str] = None,
+    litellm_params: Optional[dict] = None,
 ) -> str:
     """Async version of get_access_token."""
     credentials = credentials or _get_credentials()
@@ -125,9 +136,26 @@ async def get_access_token_async(
             status_code=401,
             message="GigaChat credentials not provided. Set GIGACHAT_CREDENTIALS or GIGACHAT_API_KEY environment variable.",
         )
+    if not litellm_params:
+        litellm_params = {}
 
-    scope = scope or _get_scope()
-    auth_url = auth_url or _get_auth_url()
+    access_token = litellm_params.get("gigachat_access_token") or get_secret_str(
+        "GIGACHAT_ACCESS_TOKEN"
+    )
+    if access_token:
+        return access_token
+
+    credentials = credentials or _get_credentials()
+    if not credentials:
+        raise GigaChatAuthError(
+            status_code=401,
+            message="GigaChat credentials not provided. Set GIGACHAT_CREDENTIALS or GIGACHAT_API_KEY environment variable.",
+        )
+
+    scope = scope or litellm_params.get("gigachat_scope") or _get_scope()
+    auth_url = (
+        auth_url or litellm_params.get("gigachat_auth_url") or _get_auth_url()
+    )
 
     # Check cache
     cache_key = f"gigachat_token:{credentials[:16]}"

litellm/llms/gigachat/chat/streaming.py

@@ -6,11 +6,12 @@ import json
 import uuid
 from typing import Any, Optional
 
+from litellm.llms.gigachat.utils import convert_usage
 from litellm.types.llms.openai import (
     ChatCompletionToolCallChunk,
     ChatCompletionToolCallFunctionChunk,
 )
-from litellm.types.utils import GenericStreamingChunk
+from litellm.types.utils import ChatCompletionUsageBlock, GenericStreamingChunk
 
 
 class GigaChatModelResponseIterator:
@@ -70,6 +71,13 @@ class GigaChatModelResponseIterator:
             )
             finish_reason = "tool_calls"
 
+        usage_block = None
+        if finish_reason == "stop":
+            usage_data = chunk.get("usage", {})
+            if usage_data:
+                usage = convert_usage(usage_data)
+                usage_block = ChatCompletionUsageBlock(**usage.dict())
+
         if finish_reason is not None:
             is_finished = True
 
@@ -78,7 +86,7 @@ class GigaChatModelResponseIterator:
             tool_use=tool_use,
             is_finished=is_finished,
             finish_reason=finish_reason or "",
-            usage=None,
+            usage=usage_block,
             index=choice.get("index", 0),
         )
 

litellm/llms/gigachat/chat/transformation.py

@@ -13,9 +13,10 @@ import httpx
 
 from litellm._logging import verbose_logger
 from litellm.llms.base_llm.chat.transformation import BaseConfig, BaseLLMException
+from litellm.llms.gigachat.utils import convert_usage, get_api_base
 from litellm.secret_managers.main import get_secret_str
 from litellm.types.llms.openai import AllMessageValues
-from litellm.types.utils import Choices, Message, ModelResponse, Usage
+from litellm.types.utils import Choices, Message, ModelResponse
 
 from ..authenticator import get_access_token
 from ..file_handler import upload_file_sync
@@ -27,9 +28,6 @@ if TYPE_CHECKING:
 else:
     LiteLLMLoggingObj = Any
 
-# GigaChat API endpoint
-GIGACHAT_BASE_URL = "https://gigachat.devices.sberbank.ru/api/v1"
-
 
 def is_valid_json(value: str) -> bool:
     """Checks whether the value passed is a valid serialized JSON string"""
@@ -94,7 +92,7 @@ class GigaChatConfig(BaseConfig):
         stream: Optional[bool] = None,
     ) -> str:
         """Get complete API URL for chat completions."""
-        base = api_base or get_secret_str("GIGACHAT_API_BASE") or GIGACHAT_BASE_URL
+        base = get_api_base(api_base)
         return f"{base}/chat/completions"
 
     def validate_environment(
@@ -116,7 +114,9 @@ class GigaChatConfig(BaseConfig):
             or get_secret_str("GIGACHAT_CREDENTIALS")
             or get_secret_str("GIGACHAT_API_KEY")
         )
-        access_token = get_access_token(credentials=credentials)
+        access_token = get_access_token(
+            credentials=credentials, litellm_params=litellm_params
+        )
 
         # Store credentials for image uploads
         self._current_credentials = credentials
@@ -467,11 +467,7 @@ class GigaChatConfig(BaseConfig):
 
         # Build usage
         usage_data = response_json.get("usage", {})
-        usage = Usage(
-            prompt_tokens=usage_data.get("prompt_tokens", 0),
-            completion_tokens=usage_data.get("completion_tokens", 0),
-            total_tokens=usage_data.get("total_tokens", 0),
-        )
+        usage = convert_usage(usage_data)
 
         model_response.id = response_json.get("id", f"chatcmpl-{uuid.uuid4().hex[:12]}")
         model_response.created = response_json.get("created", int(time.time()))

litellm/llms/gigachat/embedding/transformation.py

@@ -14,14 +14,12 @@ from litellm import LlmProviders
 from litellm.llms.base_llm.chat.transformation import BaseLLMException
 from litellm.llms.base_llm.embedding.transformation import BaseEmbeddingConfig
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
+from litellm.llms.gigachat.utils import get_api_base
 from litellm.types.llms.openai import AllEmbeddingInputValues, AllMessageValues
 from litellm.types.utils import EmbeddingResponse
 
 from ..authenticator import get_access_token
 
-# GigaChat API endpoint
-GIGACHAT_BASE_URL = "https://gigachat.devices.sberbank.ru/api/v1"
-
 
 class GigaChatEmbeddingError(BaseLLMException):
     """GigaChat Embedding API error."""
@@ -82,7 +80,7 @@ class GigaChatEmbeddingConfig(BaseEmbeddingConfig):
         Returns:
             Tuple of (custom_llm_provider, api_base, dynamic_api_key)
         """
-        api_base = api_base or GIGACHAT_BASE_URL
+        api_base = get_api_base(api_base)
         return LlmProviders.GIGACHAT.value, api_base, api_key
 
     def get_complete_url(
@@ -95,7 +93,7 @@ class GigaChatEmbeddingConfig(BaseEmbeddingConfig):
         stream: Optional[bool] = None,
     ) -> str:
         """Get the complete URL for embeddings endpoint."""
-        base = api_base or GIGACHAT_BASE_URL
+        base = get_api_base(api_base)
         return f"{base}/embeddings"
 
     def transform_embedding_request(
@@ -194,7 +192,9 @@ class GigaChatEmbeddingConfig(BaseEmbeddingConfig):
         Set up headers with OAuth token for GigaChat.
         """
         # Get access token via OAuth
-        access_token = get_access_token(api_key)
+        access_token = get_access_token(
+            credentials=api_key, litellm_params=litellm_params
+        )
 
         default_headers = {
             "Content-Type": "application/json",

litellm/llms/gigachat/file_handler.py

@@ -16,13 +16,11 @@ from litellm.llms.custom_httpx.http_handler import (
     _get_httpx_client,
     get_async_httpx_client,
 )
+from litellm.llms.gigachat.utils import get_api_base
 from litellm.types.utils import LlmProviders
 
 from .authenticator import get_access_token, get_access_token_async
 
-# GigaChat API endpoint
-GIGACHAT_BASE_URL = "https://gigachat.devices.sberbank.ru/api/v1"
-
 # Simple in-memory cache for file IDs
 _file_cache: Dict[str, str] = {}
 
@@ -82,6 +80,7 @@ def upload_file_sync(
     image_url: str,
     credentials: Optional[str] = None,
     api_base: Optional[str] = None,
+    litellm_params: Optional[dict] = None,
 ) -> Optional[str]:
     """
     Upload file to GigaChat and return file_id (sync).
@@ -114,10 +113,12 @@ def upload_file_sync(
         filename = f"{uuid.uuid4()}.{ext}"
 
         # Get access token
-        access_token = get_access_token(credentials)
+        access_token = get_access_token(
+            credentials=credentials, litellm_params=litellm_params
+        )
 
         # Upload to GigaChat
-        base_url = api_base or GIGACHAT_BASE_URL
+        base_url = get_api_base(api_base)
         upload_url = f"{base_url}/files"
 
         client = _get_httpx_client(params={"ssl_verify": False})
@@ -147,6 +148,7 @@ async def upload_file_async(
     image_url: str,
     credentials: Optional[str] = None,
     api_base: Optional[str] = None,
+    litellm_params: Optional[dict] = None,
 ) -> Optional[str]:
     """
     Upload file to GigaChat and return file_id (async).
@@ -179,10 +181,12 @@ async def upload_file_async(
         filename = f"{uuid.uuid4()}.{ext}"
 
         # Get access token
-        access_token = await get_access_token_async(credentials)
+        access_token = await get_access_token_async(
+            credentials=credentials, litellm_params=litellm_params
+        )
 
         # Upload to GigaChat
-        base_url = api_base or GIGACHAT_BASE_URL
+        base_url = get_api_base(api_base)
         upload_url = f"{base_url}/files"
 
         client = get_async_httpx_client(

litellm/llms/gigachat/utils.py

@@ -0,0 +1,35 @@
+from typing import Optional
+
+from litellm.secret_managers.main import get_secret_str
+from litellm.types.utils import PromptTokensDetailsWrapper, Usage
+
+# GigaChat API endpoint
+GIGACHAT_BASE_URL = "https://gigachat.devices.sberbank.ru/api/v1"
+
+
+def convert_usage(usage_data: dict[str, int]) -> Usage:
+    prompt_tokens = usage_data.get("prompt_tokens", 0)
+    completion_tokens = usage_data.get("completion_tokens", 0)
+    precached_prompt_tokens = usage_data.get("precached_prompt_tokens", 0)
+    total_tokens = usage_data.get("total_tokens", 0)
+
+    prompt_tokens += precached_prompt_tokens
+    total_tokens += precached_prompt_tokens
+
+    prompt_tokens_details = None
+    if precached_prompt_tokens > 0:
+        prompt_tokens_details = PromptTokensDetailsWrapper(
+            cached_tokens=precached_prompt_tokens
+        )
+
+    return Usage(
+        prompt_tokens=prompt_tokens,
+        completion_tokens=completion_tokens,
+        prompt_tokens_details=prompt_tokens_details,
+        total_tokens=total_tokens,
+    )
+
+
+@staticmethod
+def get_api_base(api_base: Optional[str] = None) -> Optional[str]:
+    return api_base or get_secret_str("GIGACHAT_API_BASE") or GIGACHAT_BASE_URL