diff --git a/docs/my-website/docs/routing.md b/docs/my-website/docs/routing.md index 4fa166548f..2b28b925f0 100644 --- a/docs/my-website/docs/routing.md +++ b/docs/my-website/docs/routing.md @@ -96,7 +96,7 @@ print(response) - `router.aimage_generation()` - async image generation calls ## Advanced - Routing Strategies -#### Routing Strategies - Weighted Pick, Rate Limit Aware, Least Busy, Latency Based +#### Routing Strategies - Weighted Pick, Rate Limit Aware, Least Busy, Latency Based, Cost Based Router provides 4 strategies for routing your calls across multiple deployments: @@ -467,6 +467,101 @@ async def router_acompletion(): asyncio.run(router_acompletion()) ``` + + + +Picks a deployment based on the lowest cost + +How this works: +- Get all healthy deployments +- Select all deployments that are under their provided `rpm/tpm` limits +- For each deployment check if `litellm_param["model"]` exists in [`litellm_model_cost_map`](https://github.com/BerriAI/litellm/blob/main/model_prices_and_context_window.json) + - if deployment does not exist in `litellm_model_cost_map` -> use deployment_cost= `$1` +- Select deployment with lowest cost + +```python +from litellm import Router +import asyncio + +model_list = [ + { + "model_name": "gpt-3.5-turbo", + "litellm_params": {"model": "gpt-4"}, + "model_info": {"id": "openai-gpt-4"}, + }, + { + "model_name": "gpt-3.5-turbo", + "litellm_params": {"model": "groq/llama3-8b-8192"}, + "model_info": {"id": "groq-llama"}, + }, +] + +# init router +router = Router(model_list=model_list, routing_strategy="cost-based-routing") +async def router_acompletion(): + response = await router.acompletion( + model="gpt-3.5-turbo", + messages=[{"role": "user", "content": "Hey, how's it going?"}] + ) + print(response) + + print(response._hidden_params["model_id"]) # expect groq-llama, since groq/llama has lowest cost + return response + +asyncio.run(router_acompletion()) + +``` + + +#### Using Custom Input/Output pricing + +Set `litellm_params["input_cost_per_token"]` and `litellm_params["output_cost_per_token"]` for using custom pricing when routing + +```python +model_list = [ + { + "model_name": "gpt-3.5-turbo", + "litellm_params": { + "model": "azure/chatgpt-v-2", + "input_cost_per_token": 0.00003, + "output_cost_per_token": 0.00003, + }, + "model_info": {"id": "chatgpt-v-experimental"}, + }, + { + "model_name": "gpt-3.5-turbo", + "litellm_params": { + "model": "azure/chatgpt-v-1", + "input_cost_per_token": 0.000000001, + "output_cost_per_token": 0.00000001, + }, + "model_info": {"id": "chatgpt-v-1"}, + }, + { + "model_name": "gpt-3.5-turbo", + "litellm_params": { + "model": "azure/chatgpt-v-5", + "input_cost_per_token": 10, + "output_cost_per_token": 12, + }, + "model_info": {"id": "chatgpt-v-5"}, + }, +] +# init router +router = Router(model_list=model_list, routing_strategy="cost-based-routing") +async def router_acompletion(): + response = await router.acompletion( + model="gpt-3.5-turbo", + messages=[{"role": "user", "content": "Hey, how's it going?"}] + ) + print(response) + + print(response._hidden_params["model_id"]) # expect chatgpt-v-1, since chatgpt-v-1 has lowest cost + return response + +asyncio.run(router_acompletion()) +``` + @@ -1159,6 +1254,7 @@ def __init__( "least-busy", "usage-based-routing", "latency-based-routing", + "cost-based-routing", ] = "simple-shuffle", ## DEBUGGING ## diff --git a/litellm/router.py b/litellm/router.py index 4353da804b..99e2435ac9 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_cost import LowestCostLoggingHandler from litellm.router_strategy.lowest_tpm_rpm_v2 import LowestTPMLoggingHandler_v2 from litellm.llms.custom_httpx.azure_dall_e_2 import ( CustomHTTPTransport, @@ -98,6 +99,7 @@ class Router: "least-busy", "usage-based-routing", "latency-based-routing", + "cost-based-routing", ] = "simple-shuffle", routing_strategy_args: dict = {}, # just for latency-based routing semaphore: Optional[asyncio.Semaphore] = None, @@ -127,7 +129,7 @@ class Router: retry_after (int): Minimum time to wait before retrying a failed request. Defaults to 0. allowed_fails (Optional[int]): Number of allowed fails before adding to cooldown. Defaults to None. cooldown_time (float): Time to cooldown a deployment after failure in seconds. Defaults to 1. - routing_strategy (Literal["simple-shuffle", "least-busy", "usage-based-routing", "latency-based-routing"]): Routing strategy. Defaults to "simple-shuffle". + routing_strategy (Literal["simple-shuffle", "least-busy", "usage-based-routing", "latency-based-routing", "cost-based-routing"]): Routing strategy. Defaults to "simple-shuffle". routing_strategy_args (dict): Additional args for latency-based routing. Defaults to {}. Returns: @@ -347,6 +349,14 @@ class Router: ) if isinstance(litellm.callbacks, list): litellm.callbacks.append(self.lowestlatency_logger) # type: ignore + elif routing_strategy == "cost-based-routing": + self.lowestcost_logger = LowestCostLoggingHandler( + router_cache=self.cache, + model_list=self.model_list, + routing_args={}, + ) + if isinstance(litellm.callbacks, list): + litellm.callbacks.append(self.lowestcost_logger) # type: ignore def print_deployment(self, deployment: dict): """ @@ -3174,6 +3184,15 @@ class Router: messages=messages, input=input, ) + elif ( + self.routing_strategy == "cost-based-routing" + and self.lowestcost_logger is not None + ): + deployment = self.lowestcost_logger.get_available_deployments( + model_group=model, + healthy_deployments=healthy_deployments, + request_kwargs=request_kwargs, + ) 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_cost.py b/litellm/router_strategy/lowest_cost.py new file mode 100644 index 0000000000..44b49378d6 --- /dev/null +++ b/litellm/router_strategy/lowest_cost.py @@ -0,0 +1,342 @@ +#### What this does #### +# picks based on response time (for streaming, this is time to first token) +from pydantic import BaseModel, Extra, Field, root_validator +import dotenv, os, requests, random +from typing import Optional, Union, List, Dict +from datetime import datetime, timedelta +import random + +dotenv.load_dotenv() # Loading env variables using dotenv +import traceback +from litellm.caching import DualCache +from litellm.integrations.custom_logger import CustomLogger +from litellm._logging import verbose_router_logger +from litellm import ModelResponse +from litellm import token_counter +import litellm + + +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 LowestCostLoggingHandler(CustomLogger): + test_flag: bool = False + logged_success: int = 0 + logged_failure: int = 0 + + def __init__( + self, router_cache: DualCache, model_list: list, routing_args: dict = {} + ): + self.router_cache = router_cache + self.model_list = model_list + + def log_success_event(self, kwargs, response_obj, start_time, end_time): + try: + """ + Update 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) + + # ------------ + # Setup values + # ------------ + """ + { + {model_group}_map: { + id: { + f"{date:hour:minute}" : {"tpm": 34, "rpm": 3} + } + } + } + """ + current_date = datetime.now().strftime("%Y-%m-%d") + current_hour = datetime.now().strftime("%H") + current_minute = datetime.now().strftime("%M") + precise_minute = f"{current_date}-{current_hour}-{current_minute}" + cost_key = f"{model_group}_map" + + response_ms: timedelta = end_time - start_time + + final_value = response_ms + total_tokens = 0 + + if isinstance(response_obj, ModelResponse): + completion_tokens = response_obj.usage.completion_tokens + total_tokens = response_obj.usage.total_tokens + final_value = float(response_ms.total_seconds() / completion_tokens) + + # ------------ + # Update usage + # ------------ + + request_count_dict = self.router_cache.get_cache(key=cost_key) or {} + + if id not in request_count_dict: + request_count_dict[id] = {} + + if precise_minute not in request_count_dict[id]: + request_count_dict[id][precise_minute] = {} + + if precise_minute not in request_count_dict[id]: + request_count_dict[id][precise_minute] = {} + + ## TPM + request_count_dict[id][precise_minute]["tpm"] = ( + request_count_dict[id][precise_minute].get("tpm", 0) + total_tokens + ) + + ## RPM + request_count_dict[id][precise_minute]["rpm"] = ( + request_count_dict[id][precise_minute].get("rpm", 0) + 1 + ) + + self.router_cache.set_cache(key=cost_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 cost 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) + + # ------------ + # Setup values + # ------------ + """ + { + {model_group}_map: { + id: { + "cost": [..] + f"{date:hour:minute}" : {"tpm": 34, "rpm": 3} + } + } + } + """ + cost_key = f"{model_group}_map" + + current_date = datetime.now().strftime("%Y-%m-%d") + current_hour = datetime.now().strftime("%H") + current_minute = datetime.now().strftime("%M") + precise_minute = f"{current_date}-{current_hour}-{current_minute}" + + response_ms: timedelta = end_time - start_time + + final_value = response_ms + total_tokens = 0 + + if isinstance(response_obj, ModelResponse): + completion_tokens = response_obj.usage.completion_tokens + total_tokens = response_obj.usage.total_tokens + final_value = float(response_ms.total_seconds() / completion_tokens) + + # ------------ + # Update usage + # ------------ + + request_count_dict = self.router_cache.get_cache(key=cost_key) or {} + + if id not in request_count_dict: + request_count_dict[id] = {} + if precise_minute not in request_count_dict[id]: + request_count_dict[id][precise_minute] = {} + + ## TPM + request_count_dict[id][precise_minute]["tpm"] = ( + request_count_dict[id][precise_minute].get("tpm", 0) + total_tokens + ) + + ## RPM + request_count_dict[id][precise_minute]["rpm"] = ( + request_count_dict[id][precise_minute].get("rpm", 0) + 1 + ) + + self.router_cache.set_cache( + key=cost_key, value=request_count_dict + ) # reset map within window + + ### TESTING ### + if self.test_flag: + self.logged_success += 1 + except Exception as e: + traceback.print_exc() + 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, + request_kwargs: Optional[Dict] = None, + ): + """ + Returns a deployment with the lowest cost + """ + cost_key = f"{model_group}_map" + + request_count_dict = self.router_cache.get_cache(key=cost_key) or {} + + # ----------------------- + # Find lowest used model + # ---------------------- + lowest_cost = float("inf") + + current_date = datetime.now().strftime("%Y-%m-%d") + current_hour = datetime.now().strftime("%H") + current_minute = datetime.now().strftime("%M") + precise_minute = f"{current_date}-{current_hour}-{current_minute}" + + deployment = None + + if request_count_dict is None: # base case + return + + all_deployments = request_count_dict + for d in healthy_deployments: + ## if healthy deployment not yet used + if d["model_info"]["id"] not in all_deployments: + all_deployments[d["model_info"]["id"]] = { + precise_minute: {"tpm": 0, "rpm": 0}, + } + + try: + input_tokens = token_counter(messages=messages, text=input) + except: + input_tokens = 0 + + # randomly sample from all_deployments, incase all deployments have latency=0.0 + _items = all_deployments.items() + + ### GET AVAILABLE DEPLOYMENTS ### filter out any deployments > tpm/rpm limits + potential_deployments = [] + _cost_per_deployment = {} + for item, item_map 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 = ( + _deployment.get("tpm", None) + or _deployment.get("litellm_params", {}).get("tpm", None) + or _deployment.get("model_info", {}).get("tpm", None) + or float("inf") + ) + + _deployment_rpm = ( + _deployment.get("rpm", None) + or _deployment.get("litellm_params", {}).get("rpm", None) + or _deployment.get("model_info", {}).get("rpm", None) + or float("inf") + ) + item_litellm_model_name = _deployment.get("litellm_params", {}).get("model") + item_litellm_model_cost_map = litellm.model_cost.get( + item_litellm_model_name, {} + ) + + # check if user provided input_cost_per_token and output_cost_per_token in litellm_params + item_input_cost = None + item_output_cost = None + if _deployment.get("litellm_params", {}).get("input_cost_per_token", None): + item_input_cost = _deployment.get("litellm_params", {}).get( + "input_cost_per_token" + ) + + if _deployment.get("litellm_params", {}).get("output_cost_per_token", None): + item_output_cost = _deployment.get("litellm_params", {}).get( + "output_cost_per_token" + ) + + if item_input_cost is None: + item_input_cost = item_litellm_model_cost_map.get( + "input_cost_per_token", 5.0 + ) + + if item_output_cost is None: + item_output_cost = item_litellm_model_cost_map.get( + "output_cost_per_token", 5.0 + ) + + # if litellm["model"] is not in model_cost map -> use item_cost = $10 + + item_cost = item_input_cost + item_output_cost + + item_rpm = item_map.get(precise_minute, {}).get("rpm", 0) + item_tpm = item_map.get(precise_minute, {}).get("tpm", 0) + + verbose_router_logger.debug( + f"item_cost: {item_cost}, item_tpm: {item_tpm}, item_rpm: {item_rpm}, model_id: {_deployment.get('model_info', {}).get('id')}" + ) + + # -------------- # + # Debugging Logic + # -------------- # + # We use _cost_per_deployment to log to langfuse, slack - this is not used to make a decision on routing + # this helps a user to debug why the router picked a specfic deployment # + _deployment_api_base = _deployment.get("litellm_params", {}).get( + "api_base", "" + ) + if _deployment_api_base is not None: + _cost_per_deployment[_deployment_api_base] = item_cost + # -------------- # + # End of Debugging Logic + # -------------- # + + if ( + item_tpm + input_tokens > _deployment_tpm + or item_rpm + 1 > _deployment_rpm + ): # if user passed in tpm / rpm in the model_list + continue + else: + potential_deployments.append((_deployment, item_cost)) + + if len(potential_deployments) == 0: + return None + + potential_deployments = sorted(potential_deployments, key=lambda x: x[1]) + + selected_deployment = potential_deployments[0][0] + return selected_deployment diff --git a/litellm/tests/test_lowest_cost_routing.py b/litellm/tests/test_lowest_cost_routing.py new file mode 100644 index 0000000000..9e627b3b46 --- /dev/null +++ b/litellm/tests/test_lowest_cost_routing.py @@ -0,0 +1,203 @@ +#### What this tests #### +# This tests the router's ability to pick deployment with lowest cost + +import sys, os, asyncio, time, random +from datetime import datetime +import traceback +from dotenv import load_dotenv + +load_dotenv() +import os, copy + +sys.path.insert( + 0, os.path.abspath("../..") +) # Adds the parent directory to the system path +import pytest +from litellm import Router +from litellm.router_strategy.lowest_cost import LowestCostLoggingHandler +from litellm.caching import DualCache + +### UNIT TESTS FOR cost ROUTING ### + + +def test_get_available_deployments(): + test_cache = DualCache() + model_list = [ + { + "model_name": "gpt-3.5-turbo", + "litellm_params": {"model": "gpt-4"}, + "model_info": {"id": "openai-gpt-4"}, + }, + { + "model_name": "gpt-3.5-turbo", + "litellm_params": {"model": "groq/llama3-8b-8192"}, + "model_info": {"id": "groq-llama"}, + }, + ] + lowest_cost_logger = LowestCostLoggingHandler( + router_cache=test_cache, model_list=model_list + ) + model_group = "gpt-3.5-turbo" + + ## CHECK WHAT'S SELECTED ## + selected_model = lowest_cost_logger.get_available_deployments( + model_group=model_group, healthy_deployments=model_list + ) + print("selected model: ", selected_model) + + assert selected_model["model_info"]["id"] == "groq-llama" + + +def test_get_available_deployments_custom_price(): + from litellm._logging import verbose_router_logger + import logging + + verbose_router_logger.setLevel(logging.DEBUG) + test_cache = DualCache() + model_list = [ + { + "model_name": "gpt-3.5-turbo", + "litellm_params": { + "model": "azure/chatgpt-v-2", + "input_cost_per_token": 0.00003, + "output_cost_per_token": 0.00003, + }, + "model_info": {"id": "chatgpt-v-experimental"}, + }, + { + "model_name": "gpt-3.5-turbo", + "litellm_params": { + "model": "azure/chatgpt-v-1", + "input_cost_per_token": 0.000000001, + "output_cost_per_token": 0.00000001, + }, + "model_info": {"id": "chatgpt-v-1"}, + }, + { + "model_name": "gpt-3.5-turbo", + "litellm_params": { + "model": "azure/chatgpt-v-5", + "input_cost_per_token": 10, + "output_cost_per_token": 12, + }, + "model_info": {"id": "chatgpt-v-5"}, + }, + ] + lowest_cost_logger = LowestCostLoggingHandler( + router_cache=test_cache, model_list=model_list + ) + model_group = "gpt-3.5-turbo" + + ## CHECK WHAT'S SELECTED ## + selected_model = lowest_cost_logger.get_available_deployments( + model_group=model_group, healthy_deployments=model_list + ) + print("selected model: ", selected_model) + + assert selected_model["model_info"]["id"] == "chatgpt-v-1" + + +@pytest.mark.asyncio +async def test_lowest_cost_routing(): + """ + Test if router returns model with the lowest cost + """ + model_list = [ + { + "model_name": "gpt-3.5-turbo", + "litellm_params": {"model": "gpt-4"}, + "model_info": {"id": "openai-gpt-4"}, + }, + { + "model_name": "gpt-3.5-turbo", + "litellm_params": {"model": "groq/llama3-8b-8192"}, + "model_info": {"id": "groq-llama"}, + }, + ] + + # init router + router = Router(model_list=model_list, routing_strategy="cost-based-routing") + response = await router.acompletion( + model="gpt-3.5-turbo", + messages=[{"role": "user", "content": "Hey, how's it going?"}], + ) + print(response) + print( + response._hidden_params["model_id"] + ) # expect groq-llama, since groq/llama has lowest cost + assert "groq-llama" == response._hidden_params["model_id"] + + +async def _deploy(lowest_cost_logger, deployment_id, tokens_used, duration): + kwargs = { + "litellm_params": { + "metadata": { + "model_group": "gpt-3.5-turbo", + "deployment": "gpt-4", + }, + "model_info": {"id": deployment_id}, + } + } + start_time = time.time() + response_obj = {"usage": {"total_tokens": tokens_used}} + time.sleep(duration) + end_time = time.time() + lowest_cost_logger.log_success_event( + response_obj=response_obj, + kwargs=kwargs, + start_time=start_time, + end_time=end_time, + ) + + +async def _gather_deploy(all_deploys): + return await asyncio.gather(*[_deploy(*t) for t in all_deploys]) + + +@pytest.mark.parametrize( + "ans_rpm", [1, 5] +) # 1 should produce nothing, 10 should select first +def test_get_available_endpoints_tpm_rpm_check_async(ans_rpm): + """ + Pass in list of 2 valid models + + Update cache with 1 model clearly being at tpm/rpm limit + + assert that only the valid model is returned + """ + from litellm._logging import verbose_router_logger + import logging + + verbose_router_logger.setLevel(logging.DEBUG) + test_cache = DualCache() + ans = "1234" + non_ans_rpm = 3 + assert ans_rpm != non_ans_rpm, "invalid test" + if ans_rpm < non_ans_rpm: + ans = None + model_list = [ + { + "model_name": "gpt-3.5-turbo", + "litellm_params": {"model": "gpt-4"}, + "model_info": {"id": "1234", "rpm": ans_rpm}, + }, + { + "model_name": "gpt-3.5-turbo", + "litellm_params": {"model": "groq/llama3-8b-8192"}, + "model_info": {"id": "5678", "rpm": non_ans_rpm}, + }, + ] + lowest_cost_logger = LowestCostLoggingHandler( + router_cache=test_cache, model_list=model_list + ) + model_group = "gpt-3.5-turbo" + d1 = [(lowest_cost_logger, "1234", 50, 0.01)] * non_ans_rpm + d2 = [(lowest_cost_logger, "5678", 50, 0.01)] * non_ans_rpm + asyncio.run(_gather_deploy([*d1, *d2])) + ## CHECK WHAT'S SELECTED ## + d_ans = lowest_cost_logger.get_available_deployments( + model_group=model_group, healthy_deployments=model_list + ) + assert (d_ans and d_ans["model_info"]["id"]) == ans + + print("selected deployment:", d_ans)