From 718f423d7daa8d55b38b66234d3a3e72c78b654b Mon Sep 17 00:00:00 2001 From: Krrish Dholakia Date: Mon, 6 May 2024 17:18:42 -0700 Subject: [PATCH 1/5] feat(slack_alerting.py): support sending daily reports on deployments allow admin to easily know slow + failing deployments Closes https://github.com/BerriAI/litellm/issues/3483 --- litellm/caching.py | 39 +++- litellm/integrations/slack_alerting.py | 282 +++++++++++++++++++++++- litellm/proxy/_super_secret_config.yaml | 2 +- litellm/proxy/utils.py | 4 + litellm/router.py | 7 +- litellm/tests/test_alerting.py | 89 +++++++- litellm/utils.py | 2 +- 7 files changed, 400 insertions(+), 25 deletions(-) diff --git a/litellm/caching.py b/litellm/caching.py index d7cf033705f..83cfe060b14 100644 --- a/litellm/caching.py +++ b/litellm/caching.py @@ -106,7 +106,7 @@ class InMemoryCache(BaseCache): return_val.append(val) return return_val - async def async_increment(self, key, value: int, **kwargs) -> int: + async def async_increment(self, key, value: float, **kwargs) -> float: # get the value init_value = await self.async_get_cache(key=key) or 0 value = init_value + value @@ -423,12 +423,12 @@ class RedisCache(BaseCache): if len(self.redis_batch_writing_buffer) >= self.redis_flush_size: await self.flush_cache_buffer() # logging done in here - async def async_increment(self, key, value: int, **kwargs) -> int: + async def async_increment(self, key, value: float, **kwargs) -> float: _redis_client = self.init_async_client() start_time = time.time() try: async with _redis_client as redis_client: - result = await redis_client.incr(name=key, amount=value) + result = await redis_client.incrbyfloat(name=key, amount=value) ## LOGGING ## end_time = time.time() _duration = end_time - start_time @@ -1382,18 +1382,41 @@ class DualCache(BaseCache): print_verbose(f"LiteLLM Cache: Excepton async add_cache: {str(e)}") traceback.print_exc() + async def async_batch_set_cache( + self, cache_list: list, local_only: bool = False, **kwargs + ): + """ + Batch write values to the cache + """ + print_verbose( + f"async batch set cache: cache keys: {cache_list}; local_only: {local_only}" + ) + try: + if self.in_memory_cache is not None: + await self.in_memory_cache.async_set_cache_pipeline( + cache_list=cache_list, **kwargs + ) + + if self.redis_cache is not None and local_only == False: + await self.redis_cache.async_set_cache_pipeline( + cache_list=cache_list, ttl=kwargs.get("ttl", None) + ) + except Exception as e: + print_verbose(f"LiteLLM Cache: Excepton async add_cache: {str(e)}") + traceback.print_exc() + async def async_increment_cache( - self, key, value: int, local_only: bool = False, **kwargs - ) -> int: + self, key, value: float, local_only: bool = False, **kwargs + ) -> float: """ Key - the key in cache - Value - int - the value you want to increment by + Value - float - the value you want to increment by - Returns - int - the incremented value + Returns - float - the incremented value """ try: - result: int = value + result: float = value if self.in_memory_cache is not None: result = await self.in_memory_cache.async_increment( key, value, **kwargs diff --git a/litellm/integrations/slack_alerting.py b/litellm/integrations/slack_alerting.py index a9aba2f1c6c..af328666f71 100644 --- a/litellm/integrations/slack_alerting.py +++ b/litellm/integrations/slack_alerting.py @@ -2,23 +2,74 @@ # Class for sending Slack Alerts # import dotenv, os +from litellm.proxy._types import UserAPIKeyAuth + dotenv.load_dotenv() # Loading env variables using dotenv -import copy -import traceback from litellm._logging import verbose_logger, verbose_proxy_logger -import litellm +import litellm, threading from typing import List, Literal, Any, Union, Optional, Dict from litellm.caching import DualCache import asyncio import aiohttp from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler import datetime +from pydantic import BaseModel +from enum import Enum +from datetime import datetime as dt, timedelta +from litellm.integrations.custom_logger import CustomLogger -class SlackAlerting: +class LiteLLMBase(BaseModel): + """ + Implements default functions, all pydantic objects should have. + """ + + def json(self, **kwargs): + try: + return self.model_dump() # noqa + except: + # if using pydantic v1 + return self.dict() + + +class SlackArgs(LiteLLMBase): + daily_report_frequency: int = 12 * 60 * 60 # 12 hours + + +class DeploymentMetrics(LiteLLMBase): + """ + Metrics per deployment, stored in cache + + Used for daily reporting + """ + + id: str + """id of deployment in router model list""" + + failed_request: bool + """did it fail the request?""" + + latency_per_output_token: Optional[float] + """latency/output token of deployment""" + + updated_at: dt + """Current time of deployment being updated""" + + +class SlackAlertingCacheKeys(Enum): + """ + Enum for deployment daily metrics keys - {deployment_id}:{enum} + """ + + failed_requests_key = "failed_requests_daily_metrics" + latency_key = "latency_daily_metrics" + + +class SlackAlerting(CustomLogger): # Class variables or attributes def __init__( self, + internal_usage_cache: DualCache, alerting_threshold: float = 300, alerting: Optional[List] = [], alert_types: Optional[ @@ -29,6 +80,7 @@ class SlackAlerting: "llm_requests_hanging", "budget_alerts", "db_exceptions", + "daily_reports", ] ] ] = [ @@ -37,6 +89,7 @@ class SlackAlerting: "llm_requests_hanging", "budget_alerts", "db_exceptions", + "daily_reports", ], alert_to_webhook_url: Optional[ Dict @@ -45,10 +98,10 @@ class SlackAlerting: self.alerting_threshold = alerting_threshold self.alerting = alerting self.alert_types = alert_types - self.internal_usage_cache = DualCache() + self.internal_usage_cache = internal_usage_cache self.async_http_handler = AsyncHTTPHandler() self.alert_to_webhook_url = alert_to_webhook_url - pass + self.is_running = False def update_values( self, @@ -196,8 +249,178 @@ class SlackAlerting: alert_type="llm_too_slow", ) - async def log_failure_event(self, original_exception: Exception): - pass + async def async_update_daily_reports( + self, deployment_metrics: DeploymentMetrics + ) -> int: + """ + Store the perf by deployment in cache + - Number of failed requests per deployment + - Latency / output tokens per deployment + + 'deployment_id:daily_metrics:failed_requests' + 'deployment_id:daily_metrics:latency_per_output_token' + + Returns + int - count of metrics set (1 - if just latency, 2 - if failed + latency) + """ + + return_val = 0 + try: + ## FAILED REQUESTS ## + if deployment_metrics.failed_request: + await self.internal_usage_cache.async_increment_cache( + key="{}:{}".format( + deployment_metrics.id, + SlackAlertingCacheKeys.failed_requests_key.value, + ), + value=1, + ) + + return_val += 1 + + ## LATENCY ## + if deployment_metrics.latency_per_output_token is not None: + await self.internal_usage_cache.async_increment_cache( + key="{}:{}".format( + deployment_metrics.id, SlackAlertingCacheKeys.latency_key.value + ), + value=deployment_metrics.latency_per_output_token, + ) + + return_val += 1 + + return return_val + except Exception as e: + return 0 + + async def send_daily_reports(self, router: litellm.Router) -> bool: + """ + Send a daily report on: + - Top 5 deployments with most failed requests + - Top 5 slowest deployments (normalized by latency/output tokens) + + Get the value from redis cache (if available) or in-memory and send it + + Cleanup: + - reset values in cache -> prevent memory leak + + Returns: + True -> if successfuly sent + False -> if not sent + """ + + ids = router.get_model_ids() + + # get keys + failed_request_keys = [ + "{}:{}".format(id, SlackAlertingCacheKeys.failed_requests_key.value) + for id in ids + ] + latency_keys = [ + "{}:{}".format(id, SlackAlertingCacheKeys.latency_key.value) for id in ids + ] + + combined_metrics_keys = failed_request_keys + latency_keys # reduce cache calls + + combined_metrics_values = await self.internal_usage_cache.async_batch_get_cache( + keys=combined_metrics_keys + ) # [1, 2, None, ..] + + all_none = True + for val in combined_metrics_values: + if val is not None: + all_none = False + + if all_none: + return False + + failed_request_values = combined_metrics_values[ + : len(failed_request_keys) + ] # # [1, 2, None, ..] + latency_values = combined_metrics_values[len(failed_request_keys) :] + + # find top 5 failed + ## Replace None values with a placeholder value (-1 in this case) + placeholder_value = 0 + replaced_failed_values = [ + value if value is not None else placeholder_value + for value in failed_request_values + ] + + ## Get the indices of top 5 keys with the highest numerical values (ignoring None values) + top_5_failed = sorted( + range(len(replaced_failed_values)), + key=lambda i: replaced_failed_values[i], + reverse=True, + )[:5] + + # find top 5 slowest + # Replace None values with a placeholder value (-1 in this case) + placeholder_value = -1 + replaced_slowest_values = [ + value if value is not None else placeholder_value + for value in latency_values + ] + + # Get the indices of top 5 values with the highest numerical values (ignoring None values) + top_5_slowest = sorted( + range(len(replaced_slowest_values)), + key=lambda i: replaced_slowest_values[i], + reverse=True, + )[:5] + + # format alert -> return the litellm model name + api base + message = f"\n\nHere are today's key metrics 📈: \n\n" + + message += "\n\n*❗️ Top 5 Deployments with Most Failed Requests:*\n\n" + for i in range(len(top_5_failed)): + key = failed_request_keys[top_5_failed[i]].split(":")[0] + _deployment = router.get_model_info(key) + if isinstance(_deployment, dict): + deployment_name = _deployment["litellm_params"].get("model", "") + else: + return False + + api_base = litellm.get_api_base( + model=deployment_name, + optional_params=( + _deployment["litellm_params"] if _deployment is not None else {} + ), + ) + if api_base is None: + api_base = "" + value = replaced_failed_values[top_5_failed[i]] + message += f"\t{i+1}. Deployment: `{deployment_name}`, Failed Requests: `{value}`, API Base: `{api_base}`\n" + + message += "\n\n*😅 Top 5 Slowest Deployments:*\n\n" + for i in range(len(top_5_slowest)): + key = latency_keys[top_5_slowest[i]].split(":")[0] + _deployment = router.get_model_info(key) + if _deployment is not None: + deployment_name = _deployment["litellm_params"].get("model", "") + else: + deployment_name = "" + api_base = litellm.get_api_base( + model=deployment_name, + optional_params=( + _deployment["litellm_params"] if _deployment is not None else {} + ), + ) + value = replaced_slowest_values[top_5_slowest[i]] + message += f"\t{i+1}. Deployment: `{deployment_name}`, Latency: `{value}`, API Base: `{api_base}`\n\n" + + # cache cleanup -> reset values to 0 + latency_cache_keys = [(key, 0) for key in latency_keys] + failed_request_cache_keys = [(key, 0) for key in failed_request_keys] + combined_metrics_cache_keys = latency_cache_keys + failed_request_cache_keys + await self.internal_usage_cache.async_batch_set_cache( + cache_list=combined_metrics_cache_keys + ) + + # send alert + await self.send_alert(message=message, level="Low", alert_type="daily_reports") + + return True async def response_taking_too_long( self, @@ -414,6 +637,7 @@ class SlackAlerting: "llm_requests_hanging", "budget_alerts", "db_exceptions", + "daily_reports", ], ): """ @@ -439,9 +663,12 @@ class SlackAlerting: # Get the current timestamp current_time = datetime.now().strftime("%H:%M:%S") _proxy_base_url = os.getenv("PROXY_BASE_URL", None) - formatted_message = ( - f"Level: `{level}`\nTimestamp: `{current_time}`\n\nMessage: {message}" - ) + if alert_type == "daily_reports": + formatted_message = message + else: + formatted_message = ( + f"Level: `{level}`\nTimestamp: `{current_time}`\n\nMessage: {message}" + ) if _proxy_base_url is not None: formatted_message += f"\n\nProxy URL: `{_proxy_base_url}`" @@ -468,3 +695,36 @@ class SlackAlerting: pass else: print("Error sending slack alert. Error=", response.text) # noqa + + async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): + """Log deployment latency""" + model_id = kwargs.get("litellm_params", {}).get("model_info", {}).get("id", "") + response_ms: timedelta = end_time - start_time + + final_value = response_ms + total_tokens = 0 + + if isinstance(response_obj, litellm.ModelResponse): + completion_tokens = response_obj.usage.completion_tokens + final_value = float(response_ms.total_seconds() / completion_tokens) + + await self.async_update_daily_reports( + DeploymentMetrics( + id=model_id, + failed_request=False, + latency_per_output_token=final_value, + updated_at=litellm.utils.get_utc_datetime(), + ) + ) + + async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time): + """Log failure + deployment latency""" + model_id = kwargs.get("litellm_params", {}).get("model_info", {}).get("id", "") + await self.async_update_daily_reports( + DeploymentMetrics( + id=model_id, + failed_request=True, + latency_per_output_token=None, + updated_at=litellm.utils.get_utc_datetime(), + ) + ) diff --git a/litellm/proxy/_super_secret_config.yaml b/litellm/proxy/_super_secret_config.yaml index d90fb13fd0f..ec8d097b145 100644 --- a/litellm/proxy/_super_secret_config.yaml +++ b/litellm/proxy/_super_secret_config.yaml @@ -19,4 +19,4 @@ litellm_settings: general_settings: alerting: ["slack"] - alert_types: ["llm_exceptions"] \ No newline at end of file + alert_types: ["llm_exceptions", "daily_reports"] \ No newline at end of file diff --git a/litellm/proxy/utils.py b/litellm/proxy/utils.py index 1048c67271b..8e66ff76ced 100644 --- a/litellm/proxy/utils.py +++ b/litellm/proxy/utils.py @@ -73,6 +73,7 @@ class ProxyLogging: "llm_requests_hanging", "budget_alerts", "db_exceptions", + "daily_reports", ] ] = [ "llm_exceptions", @@ -80,11 +81,13 @@ class ProxyLogging: "llm_requests_hanging", "budget_alerts", "db_exceptions", + "daily_reports", ] self.slack_alerting_instance = SlackAlerting( alerting_threshold=self.alerting_threshold, alerting=self.alerting, alert_types=self.alert_types, + internal_usage_cache=self.internal_usage_cache, ) def update_values( @@ -100,6 +103,7 @@ class ProxyLogging: "llm_requests_hanging", "budget_alerts", "db_exceptions", + "daily_reports", ] ] ] = None, diff --git a/litellm/router.py b/litellm/router.py index fbb245a3d93..52f778f0b45 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -2597,7 +2597,10 @@ class Router: return model return None - def get_model_ids(self): + def get_model_ids(self) -> List[str]: + """ + Returns list of model id's. + """ ids = [] for model in self.model_list: if "model_info" in model and "id" in model["model_info"]: @@ -2605,7 +2608,7 @@ class Router: ids.append(id) return ids - def get_model_names(self): + def get_model_names(self) -> List[str]: return self.model_names def get_model_list(self): diff --git a/litellm/tests/test_alerting.py b/litellm/tests/test_alerting.py index 06c4e84b316..dd77a0d3135 100644 --- a/litellm/tests/test_alerting.py +++ b/litellm/tests/test_alerting.py @@ -17,7 +17,7 @@ import asyncio from unittest.mock import patch, MagicMock from litellm.utils import get_api_base from litellm.caching import DualCache -from litellm.integrations.slack_alerting import SlackAlerting +from litellm.integrations.slack_alerting import SlackAlerting, DeploymentMetrics @pytest.mark.parametrize( @@ -116,7 +116,7 @@ from datetime import datetime, timedelta @pytest.fixture def slack_alerting(): - return SlackAlerting(alerting_threshold=1) + return SlackAlerting(alerting_threshold=1, internal_usage_cache=DualCache()) # Test for hanging LLM responses @@ -185,3 +185,88 @@ async def test_send_alert(slack_alerting): mock_post.return_value.status_code = 200 await slack_alerting.send_alert("Test message", "Low", "budget_alerts") mock_post.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_daily_reports_unit_test(slack_alerting): + with patch.object(slack_alerting, "send_alert", new=AsyncMock()) as mock_send_alert: + router = litellm.Router( + model_list=[ + { + "model_name": "test-gpt", + "litellm_params": {"model": "gpt-3.5-turbo"}, + "model_info": {"id": "1234"}, + } + ] + ) + deployment_metrics = DeploymentMetrics( + id="1234", + failed_request=False, + latency_per_output_token=20.3, + updated_at=litellm.utils.get_utc_datetime(), + ) + + updated_val = await slack_alerting.async_update_daily_reports( + deployment_metrics=deployment_metrics + ) + + assert updated_val == 1 + + await slack_alerting.send_daily_reports(router=router) + + mock_send_alert.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_daily_reports_completion(slack_alerting): + with patch.object(slack_alerting, "send_alert", new=AsyncMock()) as mock_send_alert: + litellm.callbacks = [slack_alerting] + + # on async success + router = litellm.Router( + model_list=[ + { + "model_name": "gpt-5", + "litellm_params": { + "model": "gpt-3.5-turbo", + }, + } + ] + ) + + await router.acompletion( + model="gpt-3.5-turbo", + messages=[{"role": "user", "content": "Hey, how's it going?"}], + ) + + await asyncio.sleep(3) + response_val = await slack_alerting.send_daily_reports(router=router) + + assert response_val == True + + mock_send_alert.assert_awaited_once() + + # on async failure + router = litellm.Router( + model_list=[ + { + "model_name": "gpt-5", + "litellm_params": {"model": "gpt-3.5-turbo", "api_key": "bad_key"}, + } + ] + ) + + try: + await router.acompletion( + model="gpt-3.5-turbo", + messages=[{"role": "user", "content": "Hey, how's it going?"}], + ) + except Exception as e: + pass + + await asyncio.sleep(3) + response_val = await slack_alerting.send_daily_reports(router=router) + + assert response_val == True + + mock_send_alert.assert_awaited() diff --git a/litellm/utils.py b/litellm/utils.py index 24ebcea9107..563a551d24a 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -5825,7 +5825,7 @@ def get_api_base(model: str, optional_params: dict) -> Optional[str]: Parameters: - model: str - the model passed to litellm.completion() - - optional_params - the additional params passed to litellm.completion - eg. api_base, api_key, etc. See `LiteLLM_Params` - https://github.com/BerriAI/litellm/blob/f09e6ba98d65e035a79f73bc069145002ceafd36/litellm/router.py#L67 + - optional_params - the 'litellm_params' in router.completion *OR* additional params passed to litellm.completion - eg. api_base, api_key, etc. See `LiteLLM_Params` - https://github.com/BerriAI/litellm/blob/f09e6ba98d65e035a79f73bc069145002ceafd36/litellm/router.py#L67 Returns: - string (api_base) or None From 6b9b4f05ba94b9f390f6eb3b43b96990074c8f04 Mon Sep 17 00:00:00 2001 From: Krrish Dholakia Date: Mon, 6 May 2024 18:25:48 -0700 Subject: [PATCH 2/5] feat(proxy_server.py): schedule slack daily report if enabled if user enabled daily_reports, send them a slack report every 12 hours --- litellm/integrations/slack_alerting.py | 116 ++++++++++++++++++------ litellm/proxy/_super_secret_config.yaml | 15 ++- litellm/proxy/proxy_server.py | 13 ++- litellm/proxy/utils.py | 5 + 4 files changed, 114 insertions(+), 35 deletions(-) diff --git a/litellm/integrations/slack_alerting.py b/litellm/integrations/slack_alerting.py index af328666f71..171b8a83473 100644 --- a/litellm/integrations/slack_alerting.py +++ b/litellm/integrations/slack_alerting.py @@ -17,6 +17,7 @@ from pydantic import BaseModel from enum import Enum from datetime import datetime as dt, timedelta from litellm.integrations.custom_logger import CustomLogger +import random class LiteLLMBase(BaseModel): @@ -32,8 +33,9 @@ class LiteLLMBase(BaseModel): return self.dict() -class SlackArgs(LiteLLMBase): +class SlackAlertingArgs(LiteLLMBase): daily_report_frequency: int = 12 * 60 * 60 # 12 hours + report_check_interval: int = 5 * 60 # 5 minutes class DeploymentMetrics(LiteLLMBase): @@ -63,6 +65,7 @@ class SlackAlertingCacheKeys(Enum): failed_requests_key = "failed_requests_daily_metrics" latency_key = "latency_daily_metrics" + report_sent_key = "daily_metrics_report_sent" class SlackAlerting(CustomLogger): @@ -94,6 +97,7 @@ class SlackAlerting(CustomLogger): alert_to_webhook_url: Optional[ Dict ] = None, # if user wants to separate alerts to diff channels + alerting_args={}, ): self.alerting_threshold = alerting_threshold self.alerting = alerting @@ -102,6 +106,7 @@ class SlackAlerting(CustomLogger): self.async_http_handler = AsyncHTTPHandler() self.alert_to_webhook_url = alert_to_webhook_url self.is_running = False + self.alerting_args = SlackAlertingArgs(**alerting_args) def update_values( self, @@ -109,6 +114,7 @@ class SlackAlerting(CustomLogger): alerting_threshold: Optional[float] = None, alert_types: Optional[List] = None, alert_to_webhook_url: Optional[Dict] = None, + alerting_args: Optional[Dict] = None, ): if alerting is not None: self.alerting = alerting @@ -116,7 +122,8 @@ class SlackAlerting(CustomLogger): self.alerting_threshold = alerting_threshold if alert_types is not None: self.alert_types = alert_types - + if alerting_args is not None: + self.alerting_args = SlackAlertingArgs(**alerting_args) if alert_to_webhook_url is not None: # update the dict if self.alert_to_webhook_url is None: @@ -356,7 +363,7 @@ class SlackAlerting(CustomLogger): # find top 5 slowest # Replace None values with a placeholder value (-1 in this case) - placeholder_value = -1 + placeholder_value = 0 replaced_slowest_values = [ value if value is not None else placeholder_value for value in latency_values @@ -406,8 +413,8 @@ class SlackAlerting(CustomLogger): _deployment["litellm_params"] if _deployment is not None else {} ), ) - value = replaced_slowest_values[top_5_slowest[i]] - message += f"\t{i+1}. Deployment: `{deployment_name}`, Latency: `{value}`, API Base: `{api_base}`\n\n" + value = round(replaced_slowest_values[top_5_slowest[i]], 3) + message += f"\t{i+1}. Deployment: `{deployment_name}`, Latency per output token: `{value}s/token`, API Base: `{api_base}`\n\n" # cache cleanup -> reset values to 0 latency_cache_keys = [(key, 0) for key in latency_keys] @@ -698,33 +705,82 @@ class SlackAlerting(CustomLogger): async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): """Log deployment latency""" - model_id = kwargs.get("litellm_params", {}).get("model_info", {}).get("id", "") - response_ms: timedelta = end_time - start_time - - final_value = response_ms - total_tokens = 0 - - if isinstance(response_obj, litellm.ModelResponse): - completion_tokens = response_obj.usage.completion_tokens - final_value = float(response_ms.total_seconds() / completion_tokens) - - await self.async_update_daily_reports( - DeploymentMetrics( - id=model_id, - failed_request=False, - latency_per_output_token=final_value, - updated_at=litellm.utils.get_utc_datetime(), + if "daily_reports" in self.alert_types: + model_id = ( + kwargs.get("litellm_params", {}).get("model_info", {}).get("id", "") + ) + response_s: timedelta = end_time - start_time + + final_value = response_s + total_tokens = 0 + + if isinstance(response_obj, litellm.ModelResponse): + completion_tokens = response_obj.usage.completion_tokens + final_value = float(response_s.total_seconds() / completion_tokens) + + await self.async_update_daily_reports( + DeploymentMetrics( + id=model_id, + failed_request=False, + latency_per_output_token=final_value, + updated_at=litellm.utils.get_utc_datetime(), + ) ) - ) async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time): """Log failure + deployment latency""" - model_id = kwargs.get("litellm_params", {}).get("model_info", {}).get("id", "") - await self.async_update_daily_reports( - DeploymentMetrics( - id=model_id, - failed_request=True, - latency_per_output_token=None, - updated_at=litellm.utils.get_utc_datetime(), + if "daily_reports" in self.alert_types: + model_id = ( + kwargs.get("litellm_params", {}).get("model_info", {}).get("id", "") ) - ) + await self.async_update_daily_reports( + DeploymentMetrics( + id=model_id, + failed_request=True, + latency_per_output_token=None, + updated_at=litellm.utils.get_utc_datetime(), + ) + ) + + async def _run_scheduled_daily_report(self, llm_router: Optional[litellm.Router]): + """ + If 'daily_reports' enabled + + Ping redis cache every 5 minutes to check if we should send the report + + If yes -> call send_daily_report() + """ + if llm_router is None or self.alert_types is None: + return + + if "daily_reports" in self.alert_types: + while True: + report_sent = await self.internal_usage_cache.async_get_cache( + key=SlackAlertingCacheKeys.report_sent_key.value + ) # None | datetime + + if report_sent is None: + await self.internal_usage_cache.async_set_cache( + key=SlackAlertingCacheKeys.report_sent_key.value, + value=litellm.utils.get_utc_datetime(), + ) + else: + # check if current time - interval >= time last sent + current_time = litellm.utils.get_utc_datetime() + delta = current_time - timedelta( + seconds=self.alerting_args.daily_report_frequency + ) + if delta >= report_sent: + # Sneak in the reporting logic here + await self.send_daily_reports(router=llm_router) + # Also, don't forget to update the report_sent time after sending the report! + await self.internal_usage_cache.async_set_cache( + key=SlackAlertingCacheKeys.report_sent_key.value, + value=litellm.utils.get_utc_datetime(), + ) + interval = random.randint( + self.alerting_args.report_check_interval - 3, + self.alerting_args.report_check_interval + 3, + ) # shuffle to prevent collisions + await asyncio.sleep(interval) + return diff --git a/litellm/proxy/_super_secret_config.yaml b/litellm/proxy/_super_secret_config.yaml index ec8d097b145..0475508e318 100644 --- a/litellm/proxy/_super_secret_config.yaml +++ b/litellm/proxy/_super_secret_config.yaml @@ -4,6 +4,16 @@ model_list: api_key: my-fake-key model: openai/my-fake-model model_name: fake-openai-endpoint +- litellm_params: + api_base: https://openai-function-calling-workers.tasslexyz.workers.dev/ + api_key: my-fake-key-2 + model: openai/my-fake-model-2 + model_name: fake-openai-endpoint +- litellm_params: + api_base: https://openai-function-calling-workers.tasslexyz.workers.dev/ + api_key: my-fake-key-3 + model: openai/my-fake-model-3 + model_name: fake-openai-endpoint router_settings: num_retries: 0 enable_pre_call_checks: true @@ -19,4 +29,7 @@ litellm_settings: general_settings: alerting: ["slack"] - alert_types: ["llm_exceptions", "daily_reports"] \ No newline at end of file + alert_types: ["llm_exceptions", "daily_reports"] + alerting_args: + daily_report_frequency: 60 # every minute + report_check_interval: 5 # every 5s \ No newline at end of file diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 4bb8dee7f9c..dbea1c57f2e 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -1900,9 +1900,6 @@ async def _run_background_health_check(): await asyncio.sleep(health_check_interval) -semaphore = asyncio.Semaphore(1) - - class ProxyConfig: """ Abstraction class on top of config loading/updating logic. Gives us one place to control all config updating logic. @@ -2377,6 +2374,7 @@ class ProxyConfig: alerting=general_settings.get("alerting", None), alerting_threshold=general_settings.get("alerting_threshold", 600), alert_types=general_settings.get("alert_types", None), + alerting_args=general_settings.get("alerting_args", None), redis_cache=redis_usage_cache, ) ### CONNECT TO DATABASE ### @@ -2501,7 +2499,7 @@ class ProxyConfig: for k, v in router_settings.items(): if k in available_args: router_params[k] = v - router = litellm.Router(**router_params, semaphore=semaphore) # type:ignore + router = litellm.Router(**router_params) # type:ignore return router, model_list, general_settings def get_model_info_with_id(self, model) -> RouterModelInfo: @@ -3273,6 +3271,13 @@ async def startup_event(): proxy_logging_obj._init_litellm_callbacks() # INITIALIZE LITELLM CALLBACKS ON SERVER STARTUP <- do this to catch any logging errors on startup, not when calls are being made + if "daily_reports" in proxy_logging_obj.slack_alerting_instance.alert_types: + asyncio.create_task( + proxy_logging_obj.slack_alerting_instance._run_scheduled_daily_report( + llm_router=llm_router + ) + ) # RUN DAILY REPORT (if scheduled) + ## JWT AUTH ## if general_settings.get("litellm_jwtauth", None) is not None: for k, v in general_settings["litellm_jwtauth"].items(): diff --git a/litellm/proxy/utils.py b/litellm/proxy/utils.py index 8e66ff76ced..9734806dfb3 100644 --- a/litellm/proxy/utils.py +++ b/litellm/proxy/utils.py @@ -107,6 +107,7 @@ class ProxyLogging: ] ] ] = None, + alerting_args: Optional[dict] = None, ): self.alerting = alerting if alerting_threshold is not None: @@ -118,8 +119,12 @@ class ProxyLogging: alerting=self.alerting, alerting_threshold=self.alerting_threshold, alert_types=self.alert_types, + alerting_args=alerting_args, ) + if "daily_reports" in self.alert_types: + litellm.callbacks.append(self.slack_alerting_instance) # type: ignore + if redis_cache is not None: self.internal_usage_cache.redis_cache = redis_cache From 3a8876b0d55f4bc0394ef5bd40b4da78a3bb4a3f Mon Sep 17 00:00:00 2001 From: Krrish Dholakia Date: Mon, 6 May 2024 18:29:55 -0700 Subject: [PATCH 3/5] refactor(slack_alerting.py): trigger new build --- litellm/integrations/slack_alerting.py | 1 - 1 file changed, 1 deletion(-) diff --git a/litellm/integrations/slack_alerting.py b/litellm/integrations/slack_alerting.py index 171b8a83473..b42f6d799b3 100644 --- a/litellm/integrations/slack_alerting.py +++ b/litellm/integrations/slack_alerting.py @@ -1,7 +1,6 @@ #### What this does #### # Class for sending Slack Alerts # import dotenv, os - from litellm.proxy._types import UserAPIKeyAuth dotenv.load_dotenv() # Loading env variables using dotenv From 72299a68941c5c14ffa880f68a88c94f054dd466 Mon Sep 17 00:00:00 2001 From: Krrish Dholakia Date: Mon, 6 May 2024 18:34:09 -0700 Subject: [PATCH 4/5] fix(slack_alerting.py): allow internal cache to be an optional param --- litellm/integrations/slack_alerting.py | 4 ++-- litellm/tests/test_alerting.py | 5 ++++- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/litellm/integrations/slack_alerting.py b/litellm/integrations/slack_alerting.py index b42f6d799b3..bf0f5c07d20 100644 --- a/litellm/integrations/slack_alerting.py +++ b/litellm/integrations/slack_alerting.py @@ -71,7 +71,7 @@ class SlackAlerting(CustomLogger): # Class variables or attributes def __init__( self, - internal_usage_cache: DualCache, + internal_usage_cache: Optional[DualCache] = None, alerting_threshold: float = 300, alerting: Optional[List] = [], alert_types: Optional[ @@ -101,7 +101,7 @@ class SlackAlerting(CustomLogger): self.alerting_threshold = alerting_threshold self.alerting = alerting self.alert_types = alert_types - self.internal_usage_cache = internal_usage_cache + self.internal_usage_cache = internal_usage_cache or DualCache() self.async_http_handler = AsyncHTTPHandler() self.alert_to_webhook_url = alert_to_webhook_url self.is_running = False diff --git a/litellm/tests/test_alerting.py b/litellm/tests/test_alerting.py index dd77a0d3135..c4b6157bcf4 100644 --- a/litellm/tests/test_alerting.py +++ b/litellm/tests/test_alerting.py @@ -98,7 +98,10 @@ def mock_env(monkeypatch): # Test the __init__ method def test_init(): slack_alerting = SlackAlerting( - alerting_threshold=32, alerting=["slack"], alert_types=["llm_exceptions"] + alerting_threshold=32, + alerting=["slack"], + alert_types=["llm_exceptions"], + internal_usage_cache=DualCache(), ) assert slack_alerting.alerting_threshold == 32 assert slack_alerting.alerting == ["slack"] From 69bcde6f6b0175cdd69bdbe0403aba4137617e83 Mon Sep 17 00:00:00 2001 From: Krrish Dholakia Date: Mon, 6 May 2024 18:49:36 -0700 Subject: [PATCH 5/5] test(test_router_fallbacks.py): fix test - making it easier to debug --- litellm/tests/test_router_fallbacks.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/litellm/tests/test_router_fallbacks.py b/litellm/tests/test_router_fallbacks.py index eb4cd3c0395..0e001eeba93 100644 --- a/litellm/tests/test_router_fallbacks.py +++ b/litellm/tests/test_router_fallbacks.py @@ -766,10 +766,10 @@ def test_ausage_based_routing_fallbacks(): load_dotenv() # Constants for TPM and RPM allocation - AZURE_FAST_RPM = 1 - AZURE_BASIC_RPM = 1 - OPENAI_RPM = 2 - ANTHROPIC_RPM = 100000 + AZURE_FAST_RPM = 0 + AZURE_BASIC_RPM = 0 + OPENAI_RPM = 0 + ANTHROPIC_RPM = 2 def get_azure_params(deployment_name: str): params = { @@ -854,7 +854,7 @@ def test_ausage_based_routing_fallbacks(): assert response._hidden_params["model_id"] == "1" # now make 100 mock requests to OpenAI - expect it to fallback to anthropic-claude-instant-1.2 - for i in range(21): + for i in range(3): response = router.completion( model="azure/gpt-4-fast", messages=messages, @@ -863,7 +863,7 @@ def test_ausage_based_routing_fallbacks(): ) print("response: ", response) print("response._hidden_params: ", response._hidden_params) - if i == 20: + if i == 2: # by the 19th call we should have hit TPM LIMIT for OpenAI, it should fallback to anthropic-claude-instant-1.2 assert response._hidden_params["model_id"] == "4"