Merge pull request #5235 from BerriAI/litellm_fix_s3_logs

fix(s3.py): fix s3 logging payload to have valid json values

litellm/integrations/s3.py

@@ -7,9 +7,11 @@ import subprocess
 import sys
 import traceback
 import uuid
+from typing import Optional
 
 import litellm
 from litellm._logging import print_verbose, verbose_logger
+from litellm.types.utils import StandardLoggingPayload
 
 
 class S3Logger:
@@ -123,29 +125,13 @@ class S3Logger:
                     else:
                         clean_metadata[key] = value
 
-            # Build the initial payload
-            payload = {
-                "id": id,
-                "call_type": call_type,
-                "cache_hit": cache_hit,
-                "startTime": start_time,
-                "endTime": end_time,
-                "model": kwargs.get("model", ""),
-                "user": kwargs.get("user", ""),
-                "modelParameters": optional_params,
-                "messages": messages,
-                "response": response_obj,
-                "usage": usage,
-                "metadata": clean_metadata,
-            }
-
             # Ensure everything in the payload is converted to str
-            for key, value in payload.items():
-                try:
-                    payload[key] = str(value)
-                except:
-                    # non blocking if it can't cast to a str
-                    pass
+            payload: Optional[StandardLoggingPayload] = kwargs.get(
+                "standard_logging_object", None
+            )
+
+            if payload is None:
+                return
 
             s3_file_name = litellm.utils.get_logging_id(start_time, payload) or ""
             s3_object_key = (

litellm/litellm_core_utils/litellm_logging.py

@@ -5,11 +5,13 @@ import copy
 import datetime
 import json
 import os
+import re
 import subprocess
 import sys
 import time
 import traceback
 import uuid
+from datetime import datetime as dt_object
 from typing import Any, Callable, Dict, List, Literal, Optional, Union
 
 from pydantic import BaseModel
@@ -33,6 +35,9 @@ from litellm.types.utils import (
     EmbeddingResponse,
     ImageResponse,
     ModelResponse,
+    StandardLoggingHiddenParams,
+    StandardLoggingMetadata,
+    StandardLoggingPayload,
     TextCompletionResponse,
     TranscriptionResponse,
 )
@@ -617,6 +622,16 @@ class Logging:
                     total_time=float_diff,
                 )
 
+            ## STANDARDIZED LOGGING PAYLOAD
+
+            self.model_call_details["standard_logging_object"] = (
+                get_standard_logging_object_payload(
+                    kwargs=self.model_call_details,
+                    init_response_obj=result,
+                    start_time=start_time,
+                    end_time=end_time,
+                )
+            )
             return start_time, end_time, result
         except Exception as e:
             raise Exception(f"[Non-Blocking] LiteLLM.Success_Call Error: {str(e)}")
@@ -2166,3 +2181,156 @@ def use_custom_pricing_for_model(litellm_params: Optional[dict]) -> bool:
             if k in SPECIAL_MODEL_INFO_PARAMS:
                 return True
     return False
+
+
+def is_valid_sha256_hash(value: str) -> bool:
+    # Check if the value is a valid SHA-256 hash (64 hexadecimal characters)
+    return bool(re.fullmatch(r"[a-fA-F0-9]{64}", value))
+
+
+def get_standard_logging_object_payload(
+    kwargs: Optional[dict],
+    init_response_obj: Any,
+    start_time: dt_object,
+    end_time: dt_object,
+) -> Optional[StandardLoggingPayload]:
+    try:
+        if kwargs is None:
+            kwargs = {}
+
+        hidden_params: Optional[dict] = None
+        if init_response_obj is None:
+            response_obj = {}
+        elif isinstance(init_response_obj, BaseModel):
+            response_obj = init_response_obj.model_dump()
+            hidden_params = getattr(init_response_obj, "_hidden_params", None)
+        else:
+            response_obj = {}
+        # standardize this function to be used across, s3, dynamoDB, langfuse logging
+        litellm_params = kwargs.get("litellm_params", {})
+        proxy_server_request = litellm_params.get("proxy_server_request") or {}
+        end_user_id = proxy_server_request.get("body", {}).get("user", None)
+        metadata = (
+            litellm_params.get("metadata", {}) or {}
+        )  # if litellm_params['metadata'] == None
+        completion_start_time = kwargs.get("completion_start_time", end_time)
+        call_type = kwargs.get("call_type")
+        cache_hit = kwargs.get("cache_hit", False)
+        usage = response_obj.get("usage", None) or {}
+        if type(usage) == litellm.Usage:
+            usage = dict(usage)
+        id = response_obj.get("id", kwargs.get("litellm_call_id"))
+
+        _model_id = metadata.get("model_info", {}).get("id", "")
+        _model_group = metadata.get("model_group", "")
+
+        request_tags = (
+            metadata.get("tags", [])
+            if isinstance(metadata.get("tags", []), list)
+            else []
+        )
+
+        # cleanup timestamps
+        if isinstance(start_time, datetime.datetime):
+            start_time_float = start_time.timestamp()
+        elif isinstance(start_time, float):
+            start_time_float = start_time
+        if isinstance(end_time, datetime.datetime):
+            end_time_float = end_time.timestamp()
+        elif isinstance(end_time, float):
+            end_time_float = end_time
+        if isinstance(completion_start_time, datetime.datetime):
+            completion_start_time_float = completion_start_time.timestamp()
+        elif isinstance(completion_start_time, float):
+            completion_start_time_float = completion_start_time
+        # clean up litellm hidden params
+        clean_hidden_params = StandardLoggingHiddenParams(
+            model_id=None,
+            cache_key=None,
+            api_base=None,
+            response_cost=None,
+            additional_headers=None,
+        )
+        if hidden_params is not None:
+            clean_hidden_params = StandardLoggingHiddenParams(
+                **{  # type: ignore
+                    key: hidden_params[key]
+                    for key in StandardLoggingHiddenParams.__annotations__.keys()
+                    if key in hidden_params
+                }
+            )
+        # clean up litellm metadata
+        clean_metadata = StandardLoggingMetadata(
+            user_api_key_hash=None,
+            user_api_key_alias=None,
+            user_api_key_team_id=None,
+            user_api_key_user_id=None,
+            user_api_key_team_alias=None,
+            spend_logs_metadata=None,
+            requester_ip_address=None,
+        )
+        if isinstance(metadata, dict):
+            # Filter the metadata dictionary to include only the specified keys
+            clean_metadata = StandardLoggingMetadata(
+                **{  # type: ignore
+                    key: metadata[key]
+                    for key in StandardLoggingMetadata.__annotations__.keys()
+                    if key in metadata
+                }
+            )
+
+            if metadata.get("user_api_key") is not None:
+                if is_valid_sha256_hash(str(metadata.get("user_api_key"))):
+                    clean_metadata["user_api_key_hash"] = metadata.get(
+                        "user_api_key"
+                    )  # this is the hash
+
+        if litellm.cache is not None:
+            cache_key = litellm.cache.get_cache_key(**kwargs)
+        else:
+            cache_key = None
+        if cache_hit is True:
+            import time
+
+            id = f"{id}_cache_hit{time.time()}"  # do not duplicate the request id
+
+        payload: StandardLoggingPayload = StandardLoggingPayload(
+            id=str(id),
+            call_type=call_type or "",
+            cache_hit=cache_hit,
+            startTime=start_time_float,
+            endTime=end_time_float,
+            completionStartTime=completion_start_time_float,
+            model=kwargs.get("model", "") or "",
+            metadata=clean_metadata,
+            cache_key=cache_key,
+            response_cost=kwargs.get("response_cost", 0),
+            total_tokens=usage.get("total_tokens", 0),
+            prompt_tokens=usage.get("prompt_tokens", 0),
+            completion_tokens=usage.get("completion_tokens", 0),
+            request_tags=request_tags,
+            end_user=end_user_id or "",
+            api_base=litellm_params.get("api_base", ""),
+            model_group=_model_group,
+            model_id=_model_id,
+            requester_ip_address=clean_metadata.get("requester_ip_address", None),
+            messages=kwargs.get("messages"),
+            response=(
+                response_obj if len(response_obj.keys()) > 0 else init_response_obj
+            ),
+            model_parameters=kwargs.get("optional_params", None),
+            hidden_params=clean_hidden_params,
+        )
+
+        verbose_logger.debug(
+            "Standard Logging: created payload - payload: %s\n\n", payload
+        )
+
+        return payload
+    except Exception as e:
+        verbose_logger.warning(
+            "Error creating standard logging object - {}\n{}".format(
+                str(e), traceback.format_exc()
+            )
+        )
+        return None

litellm/proxy/_new_secret_config.yaml

@@ -3,4 +3,12 @@ model_list:
     litellm_params:
       model: "gpt-4"
     model_info:
-      my_custom_key: "my_custom_value"
+      my_custom_key: "my_custom_value"
+
+litellm_settings:
+  success_callback: ["s3"]
+  s3_callback_params:
+    s3_bucket_name: mytestbucketlitellm   # AWS Bucket Name for S3
+    s3_region_name: us-west-2              # AWS Region Name for S3
+    s3_aws_access_key_id: os.environ/AWS_ACCESS_KEY_ID  # us os.environ/<variable name> to pass environment variables. This is AWS Access Key ID for S3
+    s3_aws_secret_access_key: os.environ/AWS_SECRET_ACCESS_KEY  # AWS Secret Access Key for S3

litellm/tests/test_custom_callback_input.py

@@ -1166,3 +1166,63 @@ def test_turn_off_message_logging():
 
     time.sleep(2)
     assert len(customHandler.errors) == 0
+
+
+##### VALID JSON ######
+
+
+def test_standard_logging_payload():
+    """
+    Ensure valid standard_logging_payload is passed for logging calls to s3
+
+    Motivation: provide a standard set of things that are logged to s3/gcs/future integrations across all llm calls
+    """
+    from litellm.types.utils import StandardLoggingPayload
+
+    # sync completion
+    customHandler = CompletionCustomHandler()
+    litellm.callbacks = [customHandler]
+
+    with patch.object(
+        customHandler, "log_success_event", new=MagicMock()
+    ) as mock_client:
+        _ = litellm.completion(
+            model="gpt-3.5-turbo",
+            messages=[{"role": "user", "content": "Hey, how's it going?"}],
+            mock_response="Going well!",
+        )
+
+        time.sleep(2)
+        mock_client.assert_called_once()
+
+        print(
+            f"mock_client_post.call_args: {mock_client.call_args.kwargs['kwargs'].keys()}"
+        )
+        assert "standard_logging_object" in mock_client.call_args.kwargs["kwargs"]
+        assert (
+            mock_client.call_args.kwargs["kwargs"]["standard_logging_object"]
+            is not None
+        )
+
+        print(mock_client.call_args.kwargs["kwargs"]["standard_logging_object"])
+
+        keys_list = list(StandardLoggingPayload.__annotations__.keys())
+
+        for k in keys_list:
+            assert (
+                k in mock_client.call_args.kwargs["kwargs"]["standard_logging_object"]
+            )
+
+        ## json serializable
+        json_str_payload = json.dumps(
+            mock_client.call_args.kwargs["kwargs"]["standard_logging_object"]
+        )
+        json.loads(json_str_payload)
+
+        ## response cost
+        assert (
+            mock_client.call_args.kwargs["kwargs"]["standard_logging_object"][
+                "response_cost"
+            ]
+            > 0
+        )

litellm/types/utils.py

@@ -1166,3 +1166,53 @@ class AdapterCompletionStreamWrapper:
             raise StopIteration
         except StopIteration:
             raise StopAsyncIteration
+
+
+class StandardLoggingMetadata(TypedDict):
+    """
+    Specific metadata k,v pairs logged to integration for easier cost tracking
+    """
+
+    user_api_key_hash: Optional[str]  # hash of the litellm virtual key used
+    user_api_key_alias: Optional[str]
+    user_api_key_team_id: Optional[str]
+    user_api_key_user_id: Optional[str]
+    user_api_key_team_alias: Optional[str]
+    spend_logs_metadata: Optional[
+        dict
+    ]  # special param to log k,v pairs to spendlogs for a call
+    requester_ip_address: Optional[str]
+
+
+class StandardLoggingHiddenParams(TypedDict):
+    model_id: Optional[str]
+    cache_key: Optional[str]
+    api_base: Optional[str]
+    response_cost: Optional[str]
+    additional_headers: Optional[dict]
+
+
+class StandardLoggingPayload(TypedDict):
+    id: str
+    call_type: str
+    response_cost: float
+    total_tokens: int
+    prompt_tokens: int
+    completion_tokens: int
+    startTime: float
+    endTime: float
+    completionStartTime: float
+    model: str
+    model_id: Optional[str]
+    model_group: Optional[str]
+    api_base: str
+    metadata: StandardLoggingMetadata
+    cache_hit: Optional[bool]
+    cache_key: Optional[str]
+    request_tags: list
+    end_user: Optional[str]
+    requester_ip_address: Optional[str]
+    messages: Optional[Union[str, list, dict]]
+    response: Optional[Union[str, list, dict]]
+    model_parameters: dict
+    hidden_params: StandardLoggingHiddenParams