From b2741933dc6ab5d75393dbc6843524c69824e979 Mon Sep 17 00:00:00 2001 From: Krrish Dholakia Date: Wed, 10 Apr 2024 13:23:56 -0700 Subject: [PATCH 1/7] fix(proxy_cli.py): don't double load the router config was causing callbacks to be instantiated twice - double couting usage in cache --- litellm/proxy/_new_secret_config.yaml | 13 ++++++------- litellm/proxy/proxy_cli.py | 7 ++++--- litellm/proxy/proxy_server.py | 1 + litellm/router.py | 1 - litellm/utils.py | 6 +----- 5 files changed, 12 insertions(+), 16 deletions(-) diff --git a/litellm/proxy/_new_secret_config.yaml b/litellm/proxy/_new_secret_config.yaml index a7578654dec..ce6d543727d 100644 --- a/litellm/proxy/_new_secret_config.yaml +++ b/litellm/proxy/_new_secret_config.yaml @@ -26,13 +26,12 @@ litellm_settings: success_callback: ["prometheus"] upperbound_key_generate_params: max_budget: os.environ/LITELLM_UPPERBOUND_KEYS_MAX_BUDGET - -# litellm_settings: -# drop_params: True -# max_budget: 800021 -# budget_duration: 30d -# # cache: true - + +router_settings: + routing_strategy: usage-based-routing + redis_host: redis-16337.c322.us-east-1-2.ec2.cloud.redislabs.com + redis_password: madeBerri@992 + redis_port: 16337 general_settings: master_key: sk-1234 diff --git a/litellm/proxy/proxy_cli.py b/litellm/proxy/proxy_cli.py index b8d7926963b..41eff1eaf2e 100644 --- a/litellm/proxy/proxy_cli.py +++ b/litellm/proxy/proxy_cli.py @@ -425,9 +425,10 @@ def run_server( ) proxy_config = ProxyConfig() - _, _, general_settings = asyncio.run( - proxy_config.load_config(router=None, config_file_path=config) - ) + _config = asyncio.run(proxy_config.get_config(config_file_path=config)) + general_settings = _config.get("general_settings", {}) + if general_settings is None: + general_settings = {} database_url = general_settings.get("database_url", None) db_connection_pool_limit = general_settings.get( "database_connection_pool_limit", 100 diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 691bb1adf36..d4ee7fd032a 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -2335,6 +2335,7 @@ class ProxyConfig: "background_health_checks", False ) health_check_interval = general_settings.get("health_check_interval", 300) + router_params: dict = { "cache_responses": litellm.cache != None, # cache if user passed in cache values diff --git a/litellm/router.py b/litellm/router.py index e3a9df3c993..78f7faeec1b 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -2374,7 +2374,6 @@ class Router: """ Returns the deployment based on routing strategy """ - # users need to explicitly call a specific deployment, by setting `specific_deployment = True` as completion()/embedding() kwarg # When this was no explicit we had several issues with fallbacks timing out if specific_deployment == True: diff --git a/litellm/utils.py b/litellm/utils.py index eee7f8202e9..b728225174e 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -236,7 +236,7 @@ class HiddenParams(OpenAIObject): class Config: extra = "allow" - protected_namespaces = () + protected_namespaces = () def get(self, key, default=None): # Custom .get() method to access attributes with a default value if the attribute doesn't exist @@ -1990,9 +1990,6 @@ class Logging: else: litellm.cache.add_cache(result, **kwargs) if isinstance(callback, CustomLogger): # custom logger class - print_verbose( - f"Running Async success callback: {callback}; self.stream: {self.stream}; async_complete_streaming_response: {self.model_call_details.get('async_complete_streaming_response', None)} result={result}" - ) if self.stream == True: if ( "async_complete_streaming_response" @@ -2376,7 +2373,6 @@ def client(original_function): if litellm.use_client or ( "use_client" in kwargs and kwargs["use_client"] == True ): - print_verbose(f"litedebugger initialized") if "lite_debugger" not in litellm.input_callback: litellm.input_callback.append("lite_debugger") if "lite_debugger" not in litellm.success_callback: From 180cf9bd5cb77da956c0311385eba199c7704f24 Mon Sep 17 00:00:00 2001 From: Krrish Dholakia Date: Wed, 10 Apr 2024 14:56:23 -0700 Subject: [PATCH 2/7] feat(lowest_tpm_rpm_v2.py): move to using redis.incr and redis.mget for getting model usage from redis makes routing work across multiple instances --- litellm/caching.py | 169 +++++++++++- litellm/proxy/_new_secret_config.yaml | 2 +- litellm/proxy/hooks/batch_redis_get.py | 2 +- litellm/router.py | 18 +- litellm/router_strategy/lowest_tpm_rpm_v2.py | 258 +++++++++++++++++++ 5 files changed, 437 insertions(+), 12 deletions(-) create mode 100644 litellm/router_strategy/lowest_tpm_rpm_v2.py diff --git a/litellm/caching.py b/litellm/caching.py index 6bf645f77bd..f569a050879 100644 --- a/litellm/caching.py +++ b/litellm/caching.py @@ -81,9 +81,29 @@ class InMemoryCache(BaseCache): return cached_response return None + def batch_get_cache(self, keys: list, **kwargs): + return_val = [] + for k in keys: + val = self.get_cache(key=k, **kwargs) + return_val.append(val) + return return_val + async def async_get_cache(self, key, **kwargs): return self.get_cache(key=key, **kwargs) + async def async_batch_get_cache(self, keys: list, **kwargs): + return_val = [] + for k in keys: + val = self.get_cache(key=k, **kwargs) + return_val.append(val) + return return_val + + async def async_increment(self, key, value: int, **kwargs): + # get the value + init_value = await self.async_get_cache(key=key) or 0 + value = init_value + value + await self.async_set_cache(key, value, **kwargs) + def flush_cache(self): self.cache_dict.clear() self.ttl_dict.clear() @@ -246,6 +266,19 @@ class RedisCache(BaseCache): if len(self.redis_batch_writing_buffer) >= self.redis_flush_size: await self.flush_cache_buffer() + async def async_increment(self, key, value: int, **kwargs): + _redis_client = self.init_async_client() + try: + async with _redis_client as redis_client: + await redis_client.incr(name=key, amount=value) + except Exception as e: + verbose_logger.error( + "LiteLLM Redis Caching: async async_increment() - Got exception from REDIS %s, Writing value=%s", + str(e), + value, + ) + traceback.print_exc() + async def flush_cache_buffer(self): print_verbose( f"flushing to redis....reached size of buffer {len(self.redis_batch_writing_buffer)}" @@ -283,6 +316,32 @@ class RedisCache(BaseCache): traceback.print_exc() logging.debug("LiteLLM Caching: get() - Got exception from REDIS: ", e) + def batch_get_cache(self, key_list) -> dict: + """ + Use Redis for bulk read operations + """ + key_value_dict = {} + try: + _keys = [] + for cache_key in key_list: + cache_key = self.check_and_fix_namespace(key=cache_key) + _keys.append(cache_key) + results = self.redis_client.mget(keys=_keys) + + # Associate the results back with their keys. + # 'results' is a list of values corresponding to the order of keys in 'key_list'. + key_value_dict = dict(zip(key_list, results)) + + decoded_results = { + k.decode("utf-8"): self._get_cache_logic(v) + for k, v in key_value_dict.items() + } + + return decoded_results + except Exception as e: + print_verbose(f"Error occurred in pipeline read - {str(e)}") + return key_value_dict + async def async_get_cache(self, key, **kwargs): _redis_client = self.init_async_client() key = self.check_and_fix_namespace(key=key) @@ -301,7 +360,7 @@ class RedisCache(BaseCache): f"LiteLLM Caching: async get() - Got exception from REDIS: {str(e)}" ) - async def async_get_cache_pipeline(self, key_list) -> dict: + async def async_batch_get_cache(self, key_list) -> dict: """ Use Redis for bulk read operations """ @@ -309,14 +368,11 @@ class RedisCache(BaseCache): key_value_dict = {} try: async with _redis_client as redis_client: - async with redis_client.pipeline(transaction=True) as pipe: - # Queue the get operations in the pipeline for all keys. - for cache_key in key_list: - cache_key = self.check_and_fix_namespace(key=cache_key) - pipe.get(cache_key) # Queue GET command in pipeline - - # Execute the pipeline and await the results. - results = await pipe.execute() + _keys = [] + for cache_key in key_list: + cache_key = self.check_and_fix_namespace(key=cache_key) + _keys.append(cache_key) + results = await redis_client.mget(keys=_keys) # Associate the results back with their keys. # 'results' is a list of values corresponding to the order of keys in 'key_list'. @@ -897,6 +953,39 @@ class DualCache(BaseCache): except Exception as e: traceback.print_exc() + def batch_get_cache(self, keys: list, local_only: bool = False, **kwargs): + try: + result = [None for _ in range(len(keys))] + if self.in_memory_cache is not None: + in_memory_result = self.in_memory_cache.batch_get_cache(keys, **kwargs) + + print_verbose(f"in_memory_result: {in_memory_result}") + if in_memory_result is not None: + result = in_memory_result + + if None in result and self.redis_cache is not None and local_only == False: + """ + - for the none values in the result + - check the redis cache + """ + sublist_keys = [ + key for key, value in zip(keys, result) if value is None + ] + # If not found in in-memory cache, try fetching from Redis + redis_result = self.redis_cache.batch_get_cache(sublist_keys, **kwargs) + if redis_result is not None: + # Update in-memory cache with the value from Redis + for key in redis_result: + self.in_memory_cache.set_cache(key, redis_result[key], **kwargs) + + for key, value in redis_result.items(): + result[sublist_keys.index(key)] = value + + print_verbose(f"async batch get cache: cache result: {result}") + return result + except Exception as e: + traceback.print_exc() + async def async_get_cache(self, key, local_only: bool = False, **kwargs): # Try to fetch from in-memory cache first try: @@ -930,6 +1019,50 @@ class DualCache(BaseCache): except Exception as e: traceback.print_exc() + async def async_batch_get_cache( + self, keys: list, local_only: bool = False, **kwargs + ): + try: + result = [None for _ in range(len(keys))] + if self.in_memory_cache is not None: + in_memory_result = await self.in_memory_cache.async_batch_get_cache( + keys, **kwargs + ) + + print_verbose(f"in_memory_result: {in_memory_result}") + if in_memory_result is not None: + result = in_memory_result + + if None in result and self.redis_cache is not None and local_only == False: + """ + - for the none values in the result + - check the redis cache + """ + sublist_keys = [ + key for key, value in zip(keys, result) if value is None + ] + # If not found in in-memory cache, try fetching from Redis + redis_result = await self.redis_cache.async_batch_get_cache( + sublist_keys, **kwargs + ) + + if redis_result is not None: + # Update in-memory cache with the value from Redis + for key in redis_result: + await self.in_memory_cache.async_set_cache( + key, redis_result[key], **kwargs + ) + + sublist_dict = dict(zip(sublist_keys, redis_result)) + + for key, value in sublist_dict.items(): + result[sublist_keys.index(key)] = value[key] + + print_verbose(f"async batch get cache: cache result: {result}") + return result + except Exception as e: + traceback.print_exc() + async def async_set_cache(self, key, value, local_only: bool = False, **kwargs): try: if self.in_memory_cache is not None: @@ -941,6 +1074,24 @@ class DualCache(BaseCache): 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 + ): + """ + Key - the key in cache + + Value - int - the value you want to increment by + """ + try: + if self.in_memory_cache is not None: + await self.in_memory_cache.async_increment(key, value, **kwargs) + + if self.redis_cache is not None and local_only == False: + await self.redis_cache.async_increment(key, value, **kwargs) + except Exception as e: + print_verbose(f"LiteLLM Cache: Excepton async add_cache: {str(e)}") + traceback.print_exc() + def flush_cache(self): if self.in_memory_cache is not None: self.in_memory_cache.flush_cache() diff --git a/litellm/proxy/_new_secret_config.yaml b/litellm/proxy/_new_secret_config.yaml index ce6d543727d..f2298ac7bbb 100644 --- a/litellm/proxy/_new_secret_config.yaml +++ b/litellm/proxy/_new_secret_config.yaml @@ -28,7 +28,7 @@ litellm_settings: max_budget: os.environ/LITELLM_UPPERBOUND_KEYS_MAX_BUDGET router_settings: - routing_strategy: usage-based-routing + routing_strategy: usage-based-routing-v2 redis_host: redis-16337.c322.us-east-1-2.ec2.cloud.redislabs.com redis_password: madeBerri@992 redis_port: 16337 diff --git a/litellm/proxy/hooks/batch_redis_get.py b/litellm/proxy/hooks/batch_redis_get.py index 71588c9d404..64541c1bffd 100644 --- a/litellm/proxy/hooks/batch_redis_get.py +++ b/litellm/proxy/hooks/batch_redis_get.py @@ -79,7 +79,7 @@ class _PROXY_BatchRedisRequests(CustomLogger): self.print_verbose(f"redis keys: {keys}") if len(keys) > 0: key_value_dict = ( - await litellm.cache.cache.async_get_cache_pipeline( + await litellm.cache.cache.async_batch_get_cache( key_list=keys ) ) diff --git a/litellm/router.py b/litellm/router.py index 78f7faeec1b..c6ac52bc8cb 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -21,6 +21,7 @@ from collections import defaultdict from litellm.router_strategy.least_busy import LeastBusyLoggingHandler from litellm.router_strategy.lowest_tpm_rpm import LowestTPMLoggingHandler from litellm.router_strategy.lowest_latency import LowestLatencyLoggingHandler +from litellm.router_strategy.lowest_tpm_rpm_v2 import LowestTPMLoggingHandler_v2 from litellm.llms.custom_httpx.azure_dall_e_2 import ( CustomHTTPTransport, AsyncCustomHTTPTransport, @@ -273,6 +274,12 @@ class Router: ) if isinstance(litellm.callbacks, list): litellm.callbacks.append(self.lowesttpm_logger) # type: ignore + elif routing_strategy == "usage-based-routing-v2": + self.lowesttpm_logger_v2 = LowestTPMLoggingHandler_v2( + router_cache=self.cache, model_list=self.model_list + ) + if isinstance(litellm.callbacks, list): + litellm.callbacks.append(self.lowesttpm_logger_v2) # type: ignore elif routing_strategy == "latency-based-routing": self.lowestlatency_logger = LowestLatencyLoggingHandler( router_cache=self.cache, @@ -2506,7 +2513,16 @@ class Router: messages=messages, input=input, ) - + elif ( + self.routing_strategy == "usage-based-routing-v2" + and self.lowesttpm_logger_v2 is not None + ): + deployment = self.lowesttpm_logger_v2.get_available_deployments( + model_group=model, + healthy_deployments=healthy_deployments, + messages=messages, + input=input, + ) if deployment is None: verbose_router_logger.info( f"get_available_deployment for model: {model}, No deployment available" diff --git a/litellm/router_strategy/lowest_tpm_rpm_v2.py b/litellm/router_strategy/lowest_tpm_rpm_v2.py new file mode 100644 index 00000000000..991fd57c14c --- /dev/null +++ b/litellm/router_strategy/lowest_tpm_rpm_v2.py @@ -0,0 +1,258 @@ +#### What this does #### +# identifies lowest tpm deployment + +import dotenv, os, requests, random +from typing import Optional, Union, List, Dict +from datetime import datetime + +dotenv.load_dotenv() # Loading env variables using dotenv +import traceback, asyncio +from litellm import token_counter +from litellm.caching import DualCache +from litellm.integrations.custom_logger import CustomLogger +from litellm._logging import verbose_router_logger +from litellm.utils import print_verbose + + +class LowestTPMLoggingHandler_v2(CustomLogger): + """ + Updated version of TPM/RPM Logging. + + Meant to work across instances. + + Caches individual models, not model_groups + + Uses batch get (redis.mget) + + Increments tpm/rpm limit using redis.incr + """ + + test_flag: bool = False + logged_success: int = 0 + logged_failure: int = 0 + default_cache_time_seconds: int = 1 * 60 * 60 # 1 hour + + def __init__(self, router_cache: DualCache, model_list: list): + self.router_cache = router_cache + self.model_list = model_list + + def log_success_event(self, kwargs, response_obj, start_time, end_time): + try: + """ + Update TPM/RPM usage on success + """ + if kwargs["litellm_params"].get("metadata") is None: + pass + else: + model_group = kwargs["litellm_params"]["metadata"].get( + "model_group", None + ) + + id = kwargs["litellm_params"].get("model_info", {}).get("id", None) + if model_group is None or id is None: + return + elif isinstance(id, int): + id = str(id) + + total_tokens = response_obj["usage"]["total_tokens"] + + # ------------ + # Setup values + # ------------ + current_minute = datetime.now().strftime("%H-%M") + tpm_key = f"{model_group}:tpm:{current_minute}" + rpm_key = f"{model_group}:rpm:{current_minute}" + + # ------------ + # Update usage + # ------------ + + ## TPM + request_count_dict = self.router_cache.get_cache(key=tpm_key) or {} + request_count_dict[id] = request_count_dict.get(id, 0) + total_tokens + + self.router_cache.set_cache(key=tpm_key, value=request_count_dict) + + ## RPM + request_count_dict = self.router_cache.get_cache(key=rpm_key) or {} + request_count_dict[id] = request_count_dict.get(id, 0) + 1 + + self.router_cache.set_cache(key=rpm_key, value=request_count_dict) + + ### TESTING ### + if self.test_flag: + self.logged_success += 1 + except Exception as e: + traceback.print_exc() + pass + + async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): + try: + """ + Update TPM/RPM usage on success + """ + if kwargs["litellm_params"].get("metadata") is None: + pass + else: + model_group = kwargs["litellm_params"]["metadata"].get( + "model_group", None + ) + + id = kwargs["litellm_params"].get("model_info", {}).get("id", None) + if model_group is None or id is None: + return + elif isinstance(id, int): + id = str(id) + + total_tokens = response_obj["usage"]["total_tokens"] + + # ------------ + # Setup values + # ------------ + current_minute = datetime.now().strftime("%H-%M") + + tpm_key = f"{id}:tpm:{current_minute}" + rpm_key = f"{id}:rpm:{current_minute}" + + # ------------ + # Update usage + # ------------ + # update cache + + ## TPM + await self.router_cache.async_increment_cache( + key=tpm_key, value=total_tokens + ) + ## RPM + await self.router_cache.async_increment_cache(key=rpm_key, value=1) + + ### TESTING ### + if self.test_flag: + self.logged_success += 1 + except Exception as e: + traceback.print_exc() + pass + + async def async_get_available_deployments( + self, + model_group: str, + healthy_deployments: list, + messages: Optional[List[Dict[str, str]]] = None, + input: Optional[Union[str, List]] = None, + ): + """ + Async implementation of get deployments. + + Reduces time to retrieve the tpm/rpm values from cache + """ + pass + + def get_available_deployments( + self, + model_group: str, + healthy_deployments: list, + messages: Optional[List[Dict[str, str]]] = None, + input: Optional[Union[str, List]] = None, + ): + """ + Returns a deployment with the lowest TPM/RPM usage. + """ + # get list of potential deployments + verbose_router_logger.debug( + f"get_available_deployments - Usage Based. model_group: {model_group}, healthy_deployments: {healthy_deployments}" + ) + + current_minute = datetime.now().strftime("%H-%M") + tpm_keys = [] + rpm_keys = [] + for m in healthy_deployments: + if isinstance(m, dict): + id = m.get("model_info", {}).get( + "id" + ) # a deployment should always have an 'id'. this is set in router.py + tpm_key = "{}:tpm:{}".format(id, current_minute) + rpm_key = "{}:rpm:{}".format(id, current_minute) + + tpm_keys.append(tpm_key) + rpm_keys.append(rpm_key) + + tpm_values = self.router_cache.batch_get_cache( + keys=tpm_keys + ) # [1, 2, None, ..] + rpm_values = self.router_cache.batch_get_cache( + keys=rpm_keys + ) # [1, 2, None, ..] + + tpm_dict = {} # {model_id: 1, ..} + for idx, key in enumerate(tpm_keys): + tpm_dict[tpm_keys[idx]] = tpm_values[idx] + + rpm_dict = {} # {model_id: 1, ..} + for idx, key in enumerate(rpm_keys): + rpm_dict[rpm_keys[idx]] = rpm_values[idx] + + try: + input_tokens = token_counter(messages=messages, text=input) + except: + input_tokens = 0 + verbose_router_logger.debug(f"input_tokens={input_tokens}") + # ----------------------- + # Find lowest used model + # ---------------------- + lowest_tpm = float("inf") + + if tpm_dict is None: # base case - none of the deployments have been used + # initialize a tpm dict with {model_id: 0} + tpm_dict = {} + for deployment in healthy_deployments: + tpm_dict[deployment["model_info"]["id"]] = 0 + else: + for d in healthy_deployments: + ## if healthy deployment not yet used + if d["model_info"]["id"] not in tpm_dict: + tpm_dict[d["model_info"]["id"]] = 0 + + all_deployments = tpm_dict + + deployment = None + for item, item_tpm in all_deployments.items(): + ## get the item from model list + _deployment = None + for m in healthy_deployments: + if item == m["model_info"]["id"]: + _deployment = m + + if _deployment is None: + continue # skip to next one + + _deployment_tpm = None + if _deployment_tpm is None: + _deployment_tpm = _deployment.get("tpm") + if _deployment_tpm is None: + _deployment_tpm = _deployment.get("litellm_params", {}).get("tpm") + if _deployment_tpm is None: + _deployment_tpm = _deployment.get("model_info", {}).get("tpm") + if _deployment_tpm is None: + _deployment_tpm = float("inf") + + _deployment_rpm = None + if _deployment_rpm is None: + _deployment_rpm = _deployment.get("rpm") + if _deployment_rpm is None: + _deployment_rpm = _deployment.get("litellm_params", {}).get("rpm") + if _deployment_rpm is None: + _deployment_rpm = _deployment.get("model_info", {}).get("rpm") + if _deployment_rpm is None: + _deployment_rpm = float("inf") + + if item_tpm + input_tokens > _deployment_tpm: + continue + elif (rpm_dict is not None and item in rpm_dict) and ( + rpm_dict[item] + 1 > _deployment_rpm + ): + continue + elif item_tpm < lowest_tpm: + lowest_tpm = item_tpm + deployment = _deployment + print_verbose("returning picked lowest tpm/rpm deployment.") + return deployment From a47a719caa3b437466d85e9940108e20d44857f7 Mon Sep 17 00:00:00 2001 From: Krrish Dholakia Date: Wed, 10 Apr 2024 15:23:57 -0700 Subject: [PATCH 3/7] fix(router.py): generate consistent model id's having the same id for a deployment, lets redis usage caching work across multiple instances --- litellm/router.py | 38 ++++++++++++++++++-- litellm/router_strategy/lowest_tpm_rpm_v2.py | 9 +++-- litellm/tests/test_router.py | 29 +++++++++++++++ litellm/types/router.py | 11 +++--- 4 files changed, 78 insertions(+), 9 deletions(-) diff --git a/litellm/router.py b/litellm/router.py index c6ac52bc8cb..b3b090cf2aa 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -11,7 +11,7 @@ import copy, httpx from datetime import datetime from typing import Dict, List, Optional, Union, Literal, Any, BinaryIO import random, threading, time, traceback, uuid -import litellm, openai +import litellm, openai, hashlib, json from litellm.caching import RedisCache, InMemoryCache, DualCache import logging, asyncio @@ -2072,6 +2072,34 @@ class Router: local_only=True, ) # cache for 1 hr + def _generate_model_id(self, model_group: str, litellm_params: dict): + """ + Helper function to consistently generate the same id for a deployment + + - create a string from all the litellm params + - hash + - use hash as id + """ + concat_str = model_group + for k, v in litellm_params.items(): + if isinstance(k, str): + concat_str += k + elif isinstance(k, dict): + concat_str += json.dumps(k) + else: + concat_str += str(k) + + if isinstance(v, str): + concat_str += v + elif isinstance(v, dict): + concat_str += json.dumps(v) + else: + concat_str += str(v) + + hash_object = hashlib.sha256(concat_str.encode()) + + return hash_object.hexdigest() + def set_model_list(self, model_list: list): original_model_list = copy.deepcopy(model_list) self.model_list = [] @@ -2087,7 +2115,13 @@ class Router: if isinstance(v, str) and v.startswith("os.environ/"): _litellm_params[k] = litellm.get_secret(v) - _model_info = model.pop("model_info", {}) + _model_info: dict = model.pop("model_info", {}) + + # check if model info has id + if "id" not in _model_info: + _id = self._generate_model_id(_model_name, _litellm_params) + _model_info["id"] = _id + deployment = Deployment( **model, model_name=_model_name, diff --git a/litellm/router_strategy/lowest_tpm_rpm_v2.py b/litellm/router_strategy/lowest_tpm_rpm_v2.py index 991fd57c14c..1e1f79efa29 100644 --- a/litellm/router_strategy/lowest_tpm_rpm_v2.py +++ b/litellm/router_strategy/lowest_tpm_rpm_v2.py @@ -3,6 +3,7 @@ import dotenv, os, requests, random from typing import Optional, Union, List, Dict +import datetime as datetime_og from datetime import datetime dotenv.load_dotenv() # Loading env variables using dotenv @@ -59,7 +60,7 @@ class LowestTPMLoggingHandler_v2(CustomLogger): # ------------ # Setup values # ------------ - current_minute = datetime.now().strftime("%H-%M") + current_minute = datetime.now(datetime_og.UTC).strftime("%H-%M") tpm_key = f"{model_group}:tpm:{current_minute}" rpm_key = f"{model_group}:rpm:{current_minute}" @@ -109,7 +110,9 @@ class LowestTPMLoggingHandler_v2(CustomLogger): # ------------ # Setup values # ------------ - current_minute = datetime.now().strftime("%H-%M") + current_minute = datetime.now(datetime_og.UTC).strftime( + "%H-%M" + ) # use the same timezone regardless of system clock tpm_key = f"{id}:tpm:{current_minute}" rpm_key = f"{id}:rpm:{current_minute}" @@ -162,7 +165,7 @@ class LowestTPMLoggingHandler_v2(CustomLogger): f"get_available_deployments - Usage Based. model_group: {model_group}, healthy_deployments: {healthy_deployments}" ) - current_minute = datetime.now().strftime("%H-%M") + current_minute = datetime.now(datetime_og.UTC).strftime("%H-%M") tpm_keys = [] rpm_keys = [] for m in healthy_deployments: diff --git a/litellm/tests/test_router.py b/litellm/tests/test_router.py index 97e058ec39e..1a6df388872 100644 --- a/litellm/tests/test_router.py +++ b/litellm/tests/test_router.py @@ -932,6 +932,35 @@ def test_openai_completion_on_router(): # test_openai_completion_on_router() +def test_consistent_model_id(): + """ + - For a given model group + litellm params, assert the model id is always the same + + Test on `_generate_model_id` + + Test on `set_model_list` + + Test on `_add_deployment` + """ + model_group = "gpt-3.5-turbo" + litellm_params = { + "model": "openai/my-fake-model", + "api_key": "my-fake-key", + "api_base": "https://openai-function-calling-workers.tasslexyz.workers.dev/", + "stream_timeout": 0.001, + } + + id1 = Router()._generate_model_id( + model_group=model_group, litellm_params=litellm_params + ) + + id2 = Router()._generate_model_id( + model_group=model_group, litellm_params=litellm_params + ) + + assert id1 == id2 + + def test_reading_keys_os_environ(): import openai diff --git a/litellm/types/router.py b/litellm/types/router.py index 920725131a0..8afd575f30e 100644 --- a/litellm/types/router.py +++ b/litellm/types/router.py @@ -13,7 +13,7 @@ class ModelConfig(BaseModel): rpm: int class Config: - protected_namespaces = () + protected_namespaces = () class RouterConfig(BaseModel): @@ -45,7 +45,8 @@ class RouterConfig(BaseModel): ] = "simple-shuffle" class Config: - protected_namespaces = () + protected_namespaces = () + class ModelInfo(BaseModel): id: Optional[ @@ -132,9 +133,11 @@ class Deployment(BaseModel): litellm_params: LiteLLM_Params model_info: ModelInfo - def __init__(self, model_info: Optional[ModelInfo] = None, **params): + def __init__(self, model_info: Optional[Union[ModelInfo, dict]] = None, **params): if model_info is None: model_info = ModelInfo() + elif isinstance(model_info, dict): + model_info = ModelInfo(**model_info) super().__init__(model_info=model_info, **params) def to_json(self, **kwargs): @@ -146,7 +149,7 @@ class Deployment(BaseModel): class Config: extra = "allow" - protected_namespaces = () + protected_namespaces = () def __contains__(self, key): # Define custom behavior for the 'in' operator From 2531701a2a7267a0e954838b1e9d77b39f1e31b3 Mon Sep 17 00:00:00 2001 From: Krrish Dholakia Date: Wed, 10 Apr 2024 16:57:01 -0700 Subject: [PATCH 4/7] fix(router.py): make get_cooldown_deployment logic async --- litellm/router.py | 143 +++++++++++++++--- litellm/router_strategy/lowest_tpm_rpm_v2.py | 144 +++++++++++++------ 2 files changed, 228 insertions(+), 59 deletions(-) diff --git a/litellm/router.py b/litellm/router.py index b3b090cf2aa..e343b71d73f 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -13,7 +13,7 @@ from typing import Dict, List, Optional, Union, Literal, Any, BinaryIO import random, threading, time, traceback, uuid import litellm, openai, hashlib, json from litellm.caching import RedisCache, InMemoryCache, DualCache - +import datetime as datetime_og import logging, asyncio import inspect, concurrent from openai import AsyncOpenAI @@ -414,7 +414,7 @@ class Router: verbose_router_logger.debug( f"Inside _acompletion()- model: {model}; kwargs: {kwargs}" ) - deployment = self.get_available_deployment( + deployment = await self.async_get_available_deployment( model=model, messages=messages, specific_deployment=kwargs.pop("specific_deployment", None), @@ -1605,7 +1605,7 @@ class Router: if deployment is None: return - current_minute = datetime.now().strftime("%H-%M") + current_minute = datetime.now(datetime_og.UTC).strftime("%H-%M") # get current fails for deployment # update the number of failed calls # if it's > allowed fails @@ -1643,6 +1643,22 @@ class Router: key=deployment, value=updated_fails, ttl=cooldown_time ) + async def _async_get_cooldown_deployments(self): + """ + Async implementation of '_get_cooldown_deployments' + """ + current_minute = datetime.now(datetime_og.UTC).strftime("%H-%M") + # get the current cooldown list for that minute + cooldown_key = f"{current_minute}:cooldown_models" + + # ---------------------- + # Return cooldown models + # ---------------------- + cooldown_models = await self.cache.async_get_cache(key=cooldown_key) or [] + + verbose_router_logger.debug(f"retrieve cooldown models: {cooldown_models}") + return cooldown_models + def _get_cooldown_deployments(self): """ Get the list of models being cooled down for this minute @@ -2405,7 +2421,7 @@ class Router: return _returned_deployments - def get_available_deployment( + def _common_checks_available_deployment( self, model: str, messages: Optional[List[Dict[str, str]]] = None, @@ -2413,10 +2429,8 @@ class Router: specific_deployment: Optional[bool] = False, ): """ - Returns the deployment based on routing strategy + Common checks for 'get_available_deployment' across sync + async call. """ - # users need to explicitly call a specific deployment, by setting `specific_deployment = True` as completion()/embedding() kwarg - # When this was no explicit we had several issues with fallbacks timing out if specific_deployment == True: # users can also specify a specific deployment name. At this point we should check if they are just trying to call a specific deployment for deployment in self.model_list: @@ -2456,6 +2470,111 @@ class Router: f"initial list of deployments: {healthy_deployments}" ) + verbose_router_logger.debug( + f"healthy deployments: length {len(healthy_deployments)} {healthy_deployments}" + ) + if len(healthy_deployments) == 0: + raise ValueError(f"No healthy deployment available, passed model={model}") + if litellm.model_alias_map and model in litellm.model_alias_map: + model = litellm.model_alias_map[ + model + ] # update the model to the actual value if an alias has been passed in + + return model, healthy_deployments + + async def async_get_available_deployment( + self, + model: str, + messages: Optional[List[Dict[str, str]]] = None, + input: Optional[Union[str, List]] = None, + specific_deployment: Optional[bool] = False, + ): + """ + Async implementation of 'get_available_deployments'. + + Allows all cache calls to be made async => 10x perf impact (8rps -> 100 rps). + """ + if ( + self.routing_strategy != "usage-based-routing-v2" + ): # prevent regressions for other routing strategies, that don't have async get available deployments implemented. + return self.get_available_deployment( + model=model, + messages=messages, + input=input, + specific_deployment=specific_deployment, + ) + model, healthy_deployments = self._common_checks_available_deployment( + model=model, + messages=messages, + input=input, + specific_deployment=specific_deployment, + ) + + # filter out the deployments currently cooling down + deployments_to_remove = [] + # cooldown_deployments is a list of model_id's cooling down, cooldown_deployments = ["16700539-b3cd-42f4-b426-6a12a1bb706a", "16700539-b3cd-42f4-b426-7899"] + cooldown_deployments = await self._async_get_cooldown_deployments() + verbose_router_logger.debug( + f"async cooldown deployments: {cooldown_deployments}" + ) + # Find deployments in model_list whose model_id is cooling down + for deployment in healthy_deployments: + deployment_id = deployment["model_info"]["id"] + if deployment_id in cooldown_deployments: + deployments_to_remove.append(deployment) + # remove unhealthy deployments from healthy deployments + for deployment in deployments_to_remove: + healthy_deployments.remove(deployment) + + # filter pre-call checks + if self.enable_pre_call_checks and messages is not None: + healthy_deployments = self._pre_call_checks( + model=model, healthy_deployments=healthy_deployments, messages=messages + ) + + if ( + self.routing_strategy == "usage-based-routing-v2" + and self.lowesttpm_logger_v2 is not None + ): + deployment = await self.lowesttpm_logger_v2.async_get_available_deployments( + model_group=model, + healthy_deployments=healthy_deployments, + messages=messages, + input=input, + ) + + if deployment is None: + verbose_router_logger.info( + f"get_available_deployment for model: {model}, No deployment available" + ) + raise ValueError( + f"No deployments available for selected model, passed model={model}" + ) + verbose_router_logger.info( + f"get_available_deployment for model: {model}, Selected deployment: {self.print_deployment(deployment)} for model: {model}" + ) + return deployment + + def get_available_deployment( + self, + model: str, + messages: Optional[List[Dict[str, str]]] = None, + input: Optional[Union[str, List]] = None, + specific_deployment: Optional[bool] = False, + ): + """ + Returns the deployment based on routing strategy + """ + # users need to explicitly call a specific deployment, by setting `specific_deployment = True` as completion()/embedding() kwarg + # When this was no explicit we had several issues with fallbacks timing out + + model, healthy_deployments = self._common_checks_available_deployment( + model=model, + messages=messages, + input=input, + specific_deployment=specific_deployment, + ) + # filter out the deployments currently cooling down deployments_to_remove = [] # cooldown_deployments is a list of model_id's cooling down, cooldown_deployments = ["16700539-b3cd-42f4-b426-6a12a1bb706a", "16700539-b3cd-42f4-b426-7899"] @@ -2476,16 +2595,6 @@ class Router: model=model, healthy_deployments=healthy_deployments, messages=messages ) - verbose_router_logger.debug( - f"healthy deployments: length {len(healthy_deployments)} {healthy_deployments}" - ) - if len(healthy_deployments) == 0: - raise ValueError(f"No healthy deployment available, passed model={model}") - if litellm.model_alias_map and model in litellm.model_alias_map: - model = litellm.model_alias_map[ - model - ] # update the model to the actual value if an alias has been passed in - if self.routing_strategy == "least-busy" and self.leastbusy_logger is not None: deployment = self.leastbusy_logger.get_available_deployments( model_group=model, healthy_deployments=healthy_deployments diff --git a/litellm/router_strategy/lowest_tpm_rpm_v2.py b/litellm/router_strategy/lowest_tpm_rpm_v2.py index 1e1f79efa29..8f9f57fd970 100644 --- a/litellm/router_strategy/lowest_tpm_rpm_v2.py +++ b/litellm/router_strategy/lowest_tpm_rpm_v2.py @@ -136,56 +136,20 @@ class LowestTPMLoggingHandler_v2(CustomLogger): traceback.print_exc() pass - async def async_get_available_deployments( + def _common_checks_available_deployment( self, model_group: str, healthy_deployments: list, + tpm_keys: list, + tpm_values: list, + rpm_keys: list, + rpm_values: list, messages: Optional[List[Dict[str, str]]] = None, input: Optional[Union[str, List]] = None, ): """ - Async implementation of get deployments. - - Reduces time to retrieve the tpm/rpm values from cache + Common checks for get available deployment, across sync + async implementations """ - pass - - def get_available_deployments( - self, - model_group: str, - healthy_deployments: list, - messages: Optional[List[Dict[str, str]]] = None, - input: Optional[Union[str, List]] = None, - ): - """ - Returns a deployment with the lowest TPM/RPM usage. - """ - # get list of potential deployments - verbose_router_logger.debug( - f"get_available_deployments - Usage Based. model_group: {model_group}, healthy_deployments: {healthy_deployments}" - ) - - current_minute = datetime.now(datetime_og.UTC).strftime("%H-%M") - tpm_keys = [] - rpm_keys = [] - for m in healthy_deployments: - if isinstance(m, dict): - id = m.get("model_info", {}).get( - "id" - ) # a deployment should always have an 'id'. this is set in router.py - tpm_key = "{}:tpm:{}".format(id, current_minute) - rpm_key = "{}:rpm:{}".format(id, current_minute) - - tpm_keys.append(tpm_key) - rpm_keys.append(rpm_key) - - tpm_values = self.router_cache.batch_get_cache( - keys=tpm_keys - ) # [1, 2, None, ..] - rpm_values = self.router_cache.batch_get_cache( - keys=rpm_keys - ) # [1, 2, None, ..] - tpm_dict = {} # {model_id: 1, ..} for idx, key in enumerate(tpm_keys): tpm_dict[tpm_keys[idx]] = tpm_values[idx] @@ -259,3 +223,99 @@ class LowestTPMLoggingHandler_v2(CustomLogger): deployment = _deployment print_verbose("returning picked lowest tpm/rpm deployment.") return deployment + + async def async_get_available_deployments( + self, + model_group: str, + healthy_deployments: list, + messages: Optional[List[Dict[str, str]]] = None, + input: Optional[Union[str, List]] = None, + ): + """ + Async implementation of get deployments. + + Reduces time to retrieve the tpm/rpm values from cache + """ + # get list of potential deployments + verbose_router_logger.debug( + f"get_available_deployments - Usage Based. model_group: {model_group}, healthy_deployments: {healthy_deployments}" + ) + + current_minute = datetime.now(datetime_og.UTC).strftime("%H-%M") + tpm_keys = [] + rpm_keys = [] + for m in healthy_deployments: + if isinstance(m, dict): + id = m.get("model_info", {}).get( + "id" + ) # a deployment should always have an 'id'. this is set in router.py + tpm_key = "{}:tpm:{}".format(id, current_minute) + rpm_key = "{}:rpm:{}".format(id, current_minute) + + tpm_keys.append(tpm_key) + rpm_keys.append(rpm_key) + + tpm_values = await self.router_cache.async_batch_get_cache( + keys=tpm_keys + ) # [1, 2, None, ..] + rpm_values = await self.router_cache.async_batch_get_cache( + keys=rpm_keys + ) # [1, 2, None, ..] + + return self._common_checks_available_deployment( + model_group=model_group, + healthy_deployments=healthy_deployments, + tpm_keys=tpm_keys, + tpm_values=tpm_values, + rpm_keys=rpm_keys, + rpm_values=rpm_values, + messages=messages, + input=input, + ) + + def get_available_deployments( + self, + model_group: str, + healthy_deployments: list, + messages: Optional[List[Dict[str, str]]] = None, + input: Optional[Union[str, List]] = None, + ): + """ + Returns a deployment with the lowest TPM/RPM usage. + """ + # get list of potential deployments + verbose_router_logger.debug( + f"get_available_deployments - Usage Based. model_group: {model_group}, healthy_deployments: {healthy_deployments}" + ) + + current_minute = datetime.now(datetime_og.UTC).strftime("%H-%M") + tpm_keys = [] + rpm_keys = [] + for m in healthy_deployments: + if isinstance(m, dict): + id = m.get("model_info", {}).get( + "id" + ) # a deployment should always have an 'id'. this is set in router.py + tpm_key = "{}:tpm:{}".format(id, current_minute) + rpm_key = "{}:rpm:{}".format(id, current_minute) + + tpm_keys.append(tpm_key) + rpm_keys.append(rpm_key) + + tpm_values = self.router_cache.batch_get_cache( + keys=tpm_keys + ) # [1, 2, None, ..] + rpm_values = self.router_cache.batch_get_cache( + keys=rpm_keys + ) # [1, 2, None, ..] + + return self._common_checks_available_deployment( + model_group=model_group, + healthy_deployments=healthy_deployments, + tpm_keys=tpm_keys, + tpm_values=tpm_values, + rpm_keys=rpm_keys, + rpm_values=rpm_values, + messages=messages, + input=input, + ) From 37ac17aebdec374549ee2feaea016acd13d43052 Mon Sep 17 00:00:00 2001 From: Krrish Dholakia Date: Wed, 10 Apr 2024 17:55:24 -0700 Subject: [PATCH 5/7] fix(router.py): fix datetime object --- litellm/router.py | 24 ++++++++++++-------- litellm/router_strategy/lowest_tpm_rpm_v2.py | 14 ++++++++---- litellm/utils.py | 10 ++++++++ 3 files changed, 33 insertions(+), 15 deletions(-) diff --git a/litellm/router.py b/litellm/router.py index e343b71d73f..ca28dcb0704 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -26,7 +26,7 @@ from litellm.llms.custom_httpx.azure_dall_e_2 import ( CustomHTTPTransport, AsyncCustomHTTPTransport, ) -from litellm.utils import ModelResponse, CustomStreamWrapper +from litellm.utils import ModelResponse, CustomStreamWrapper, get_utc_datetime import copy from litellm._logging import verbose_router_logger import logging @@ -588,7 +588,7 @@ class Router: verbose_router_logger.debug( f"Inside _image_generation()- model: {model}; kwargs: {kwargs}" ) - deployment = self.get_available_deployment( + deployment = await self.async_get_available_deployment( model=model, messages=[{"role": "user", "content": "prompt"}], specific_deployment=kwargs.pop("specific_deployment", None), @@ -688,7 +688,7 @@ class Router: verbose_router_logger.debug( f"Inside _atranscription()- model: {model}; kwargs: {kwargs}" ) - deployment = self.get_available_deployment( + deployment = await self.async_get_available_deployment( model=model, messages=[{"role": "user", "content": "prompt"}], specific_deployment=kwargs.pop("specific_deployment", None), @@ -768,7 +768,7 @@ class Router: verbose_router_logger.debug( f"Inside _moderation()- model: {model}; kwargs: {kwargs}" ) - deployment = self.get_available_deployment( + deployment = await self.async_get_available_deployment( model=model, input=input, specific_deployment=kwargs.pop("specific_deployment", None), @@ -911,7 +911,7 @@ class Router: verbose_router_logger.debug( f"Inside _atext_completion()- model: {model}; kwargs: {kwargs}" ) - deployment = self.get_available_deployment( + deployment = await self.async_get_available_deployment( model=model, messages=[{"role": "user", "content": prompt}], specific_deployment=kwargs.pop("specific_deployment", None), @@ -1077,7 +1077,7 @@ class Router: verbose_router_logger.debug( f"Inside _aembedding()- model: {model}; kwargs: {kwargs}" ) - deployment = self.get_available_deployment( + deployment = await self.async_get_available_deployment( model=model, input=input, specific_deployment=kwargs.pop("specific_deployment", None), @@ -1605,7 +1605,8 @@ class Router: if deployment is None: return - current_minute = datetime.now(datetime_og.UTC).strftime("%H-%M") + dt = get_utc_datetime() + current_minute = dt.strftime("%H-%M") # get current fails for deployment # update the number of failed calls # if it's > allowed fails @@ -1647,7 +1648,8 @@ class Router: """ Async implementation of '_get_cooldown_deployments' """ - current_minute = datetime.now(datetime_og.UTC).strftime("%H-%M") + dt = get_utc_datetime() + current_minute = dt.strftime("%H-%M") # get the current cooldown list for that minute cooldown_key = f"{current_minute}:cooldown_models" @@ -1663,7 +1665,8 @@ class Router: """ Get the list of models being cooled down for this minute """ - current_minute = datetime.now().strftime("%H-%M") + dt = get_utc_datetime() + current_minute = dt.strftime("%H-%M") # get the current cooldown list for that minute cooldown_key = f"{current_minute}:cooldown_models" @@ -2336,7 +2339,8 @@ class Router: _rate_limit_error = False ## get model group RPM ## - current_minute = datetime.now().strftime("%H-%M") + dt = get_utc_datetime() + current_minute = dt.strftime("%H-%M") rpm_key = f"{model}:rpm:{current_minute}" model_group_cache = ( self.cache.get_cache(key=rpm_key, local_only=True) or {} diff --git a/litellm/router_strategy/lowest_tpm_rpm_v2.py b/litellm/router_strategy/lowest_tpm_rpm_v2.py index 8f9f57fd970..c5598c11eb9 100644 --- a/litellm/router_strategy/lowest_tpm_rpm_v2.py +++ b/litellm/router_strategy/lowest_tpm_rpm_v2.py @@ -12,7 +12,7 @@ from litellm import token_counter from litellm.caching import DualCache from litellm.integrations.custom_logger import CustomLogger from litellm._logging import verbose_router_logger -from litellm.utils import print_verbose +from litellm.utils import print_verbose, get_utc_datetime class LowestTPMLoggingHandler_v2(CustomLogger): @@ -60,7 +60,8 @@ class LowestTPMLoggingHandler_v2(CustomLogger): # ------------ # Setup values # ------------ - current_minute = datetime.now(datetime_og.UTC).strftime("%H-%M") + dt = get_utc_datetime() + current_minute = dt.strftime("%H-%M") tpm_key = f"{model_group}:tpm:{current_minute}" rpm_key = f"{model_group}:rpm:{current_minute}" @@ -110,7 +111,8 @@ class LowestTPMLoggingHandler_v2(CustomLogger): # ------------ # Setup values # ------------ - current_minute = datetime.now(datetime_og.UTC).strftime( + dt = get_utc_datetime() + current_minute = dt.strftime( "%H-%M" ) # use the same timezone regardless of system clock @@ -241,7 +243,8 @@ class LowestTPMLoggingHandler_v2(CustomLogger): f"get_available_deployments - Usage Based. model_group: {model_group}, healthy_deployments: {healthy_deployments}" ) - current_minute = datetime.now(datetime_og.UTC).strftime("%H-%M") + dt = get_utc_datetime() + current_minute = dt.strftime("%H-%M") tpm_keys = [] rpm_keys = [] for m in healthy_deployments: @@ -288,7 +291,8 @@ class LowestTPMLoggingHandler_v2(CustomLogger): f"get_available_deployments - Usage Based. model_group: {model_group}, healthy_deployments: {healthy_deployments}" ) - current_minute = datetime.now(datetime_og.UTC).strftime("%H-%M") + dt = get_utc_datetime() + current_minute = dt.strftime("%H-%M") tpm_keys = [] rpm_keys = [] for m in healthy_deployments: diff --git a/litellm/utils.py b/litellm/utils.py index b728225174e..5c5f816c860 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -5908,6 +5908,16 @@ def get_api_key(llm_provider: str, dynamic_api_key: Optional[str]): return api_key +def get_utc_datetime(): + import datetime as dt + from datetime import datetime + + if hasattr(dt, "UTC"): + return datetime.now(dt.UTC) # type: ignore + else: + return datetime.utcnow() # type: ignore + + def get_max_tokens(model: str): """ Get the maximum number of output tokens allowed for a given model. From 52462e8bac7b2ce5364f85044b26be9bfed8c2ab Mon Sep 17 00:00:00 2001 From: Krrish Dholakia Date: Wed, 10 Apr 2024 18:06:31 -0700 Subject: [PATCH 6/7] fix(router.py): move specific deployment check outside common functions --- litellm/router.py | 36 ++++++++++++++++++++++++------------ 1 file changed, 24 insertions(+), 12 deletions(-) diff --git a/litellm/router.py b/litellm/router.py index ca28dcb0704..ad3129304d8 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -2435,18 +2435,6 @@ class Router: """ Common checks for 'get_available_deployment' across sync + async call. """ - if specific_deployment == True: - # users can also specify a specific deployment name. At this point we should check if they are just trying to call a specific deployment - for deployment in self.model_list: - deployment_model = deployment.get("litellm_params").get("model") - if deployment_model == model: - # User Passed a specific deployment name on their config.yaml, example azure/chat-gpt-v-2 - # return the first deployment where the `model` matches the specificed deployment name - return deployment - raise ValueError( - f"LiteLLM Router: Trying to call specific deployment, but Model:{model} does not exist in Model List: {self.model_list}" - ) - # check if aliases set on litellm model alias map if model in self.model_group_alias: verbose_router_logger.debug( @@ -2507,6 +2495,19 @@ class Router: input=input, specific_deployment=specific_deployment, ) + + if specific_deployment == True: + # users can also specify a specific deployment name. At this point we should check if they are just trying to call a specific deployment + for deployment in self.model_list: + deployment_model = deployment.get("litellm_params").get("model") + if deployment_model == model: + # User Passed a specific deployment name on their config.yaml, example azure/chat-gpt-v-2 + # return the first deployment where the `model` matches the specificed deployment name + return deployment + raise ValueError( + f"LiteLLM Router: Trying to call specific deployment, but Model:{model} does not exist in Model List: {self.model_list}" + ) + model, healthy_deployments = self._common_checks_available_deployment( model=model, messages=messages, @@ -2571,6 +2572,17 @@ class Router: """ # users need to explicitly call a specific deployment, by setting `specific_deployment = True` as completion()/embedding() kwarg # When this was no explicit we had several issues with fallbacks timing out + if specific_deployment == True: + # users can also specify a specific deployment name. At this point we should check if they are just trying to call a specific deployment + for deployment in self.model_list: + deployment_model = deployment.get("litellm_params").get("model") + if deployment_model == model: + # User Passed a specific deployment name on their config.yaml, example azure/chat-gpt-v-2 + # return the first deployment where the `model` matches the specificed deployment name + return deployment + raise ValueError( + f"LiteLLM Router: Trying to call specific deployment, but Model:{model} does not exist in Model List: {self.model_list}" + ) model, healthy_deployments = self._common_checks_available_deployment( model=model, From 266dba65e73875b188269001cf1d2690813cf01c Mon Sep 17 00:00:00 2001 From: Krrish Dholakia Date: Wed, 10 Apr 2024 18:32:54 -0700 Subject: [PATCH 7/7] fix(router.py): handle 1 deployment being picked --- litellm/router.py | 45 +++++++++++++++++++++------------------------ 1 file changed, 21 insertions(+), 24 deletions(-) diff --git a/litellm/router.py b/litellm/router.py index ad3129304d8..072662baee4 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -2434,8 +2434,22 @@ class Router: ): """ Common checks for 'get_available_deployment' across sync + async call. + + If 'healthy_deployments' returned is None, this means the user chose a specific deployment """ # check if aliases set on litellm model alias map + if specific_deployment == True: + # users can also specify a specific deployment name. At this point we should check if they are just trying to call a specific deployment + for deployment in self.model_list: + deployment_model = deployment.get("litellm_params").get("model") + if deployment_model == model: + # User Passed a specific deployment name on their config.yaml, example azure/chat-gpt-v-2 + # return the first deployment where the `model` matches the specificed deployment name + return deployment, None + raise ValueError( + f"LiteLLM Router: Trying to call specific deployment, but Model:{model} does not exist in Model List: {self.model_list}" + ) + if model in self.model_group_alias: verbose_router_logger.debug( f"Using a model alias. Got Request for {model}, sending requests to {self.model_group_alias.get(model)}" @@ -2447,7 +2461,7 @@ class Router: self.default_deployment ) # self.default_deployment updated_deployment["litellm_params"]["model"] = model - return updated_deployment + return updated_deployment, None ## get healthy deployments ### get all deployments @@ -2496,18 +2510,6 @@ class Router: specific_deployment=specific_deployment, ) - if specific_deployment == True: - # users can also specify a specific deployment name. At this point we should check if they are just trying to call a specific deployment - for deployment in self.model_list: - deployment_model = deployment.get("litellm_params").get("model") - if deployment_model == model: - # User Passed a specific deployment name on their config.yaml, example azure/chat-gpt-v-2 - # return the first deployment where the `model` matches the specificed deployment name - return deployment - raise ValueError( - f"LiteLLM Router: Trying to call specific deployment, but Model:{model} does not exist in Model List: {self.model_list}" - ) - model, healthy_deployments = self._common_checks_available_deployment( model=model, messages=messages, @@ -2515,6 +2517,9 @@ class Router: specific_deployment=specific_deployment, ) + if healthy_deployments is None: + return model + # filter out the deployments currently cooling down deployments_to_remove = [] # cooldown_deployments is a list of model_id's cooling down, cooldown_deployments = ["16700539-b3cd-42f4-b426-6a12a1bb706a", "16700539-b3cd-42f4-b426-7899"] @@ -2572,17 +2577,6 @@ class Router: """ # users need to explicitly call a specific deployment, by setting `specific_deployment = True` as completion()/embedding() kwarg # When this was no explicit we had several issues with fallbacks timing out - if specific_deployment == True: - # users can also specify a specific deployment name. At this point we should check if they are just trying to call a specific deployment - for deployment in self.model_list: - deployment_model = deployment.get("litellm_params").get("model") - if deployment_model == model: - # User Passed a specific deployment name on their config.yaml, example azure/chat-gpt-v-2 - # return the first deployment where the `model` matches the specificed deployment name - return deployment - raise ValueError( - f"LiteLLM Router: Trying to call specific deployment, but Model:{model} does not exist in Model List: {self.model_list}" - ) model, healthy_deployments = self._common_checks_available_deployment( model=model, @@ -2591,6 +2585,9 @@ class Router: specific_deployment=specific_deployment, ) + if healthy_deployments is None: + return model + # filter out the deployments currently cooling down deployments_to_remove = [] # cooldown_deployments is a list of model_id's cooling down, cooldown_deployments = ["16700539-b3cd-42f4-b426-6a12a1bb706a", "16700539-b3cd-42f4-b426-7899"]