diff --git a/docs/my-website/docs/load_test.md b/docs/my-website/docs/load_test.md index bd97c62290b..c1fac60fcb2 100644 --- a/docs/my-website/docs/load_test.md +++ b/docs/my-website/docs/load_test.md @@ -213,7 +213,191 @@ asyncio.run(loadtest_fn()) ``` -## Multi-Instance TPM/RPM Load Test +## Multi-Instance TPM/RPM Load Test (Router) + +Test if your defined tpm/rpm limits are respected across multiple instances of the Router object. + +In our test: +- Max RPM per deployment is = 100 requests per minute +- Max Throughput / min on router = 200 requests per minute (2 deployments) +- Load we'll send through router = 600 requests per minute + +:::info + +If you don't want to call a real LLM API endpoint, you can setup a fake openai server. [See code](#extra---setup-fake-openai-server) + +::: + +### 1. Setup Router + +```python +from litellm import Router +import litellm +litellm.suppress_debug_info = True +litellm.set_verbose = False +import logging +logging.basicConfig(level=logging.CRITICAL) +import os, random, uuid, time, asyncio + +os.environ["REDIS_SSL"] = "True" + +# Model list for OpenAI and Anthropic models +model_list = [ + { + "model_name": "fake-openai-endpoint", + "litellm_params": { + "model": "gpt-3.5-turbo", + "api_key": "my-fake-key", + "api_base": "http://0.0.0.0:8080", + "rpm": 100 + }, + }, + { + "model_name": "fake-openai-endpoint", + "litellm_params": { + "model": "gpt-3.5-turbo", + "api_key": "my-fake-key", + "api_base": "http://0.0.0.0:8081", + "rpm": 100 + }, + }, +] + +router_1 = Router(model_list=model_list, num_retries=0, enable_pre_call_checks=True, redis_host=os.getenv("REDIS_HOST"), redis_port=os.getenv("REDIS_PORT"), redis_password=os.getenv("REDIS_PASSWORD")) +router_2 = Router(model_list=model_list, num_retries=0, enable_pre_call_checks=True, redis_host=os.getenv("REDIS_HOST"), redis_port=os.getenv("REDIS_PORT"), redis_password=os.getenv("REDIS_PASSWORD")) + + + +async def router_completion_non_streaming(): + try: + client: Router = random.sample([router_1, router_2], 1)[0] # randomly pick b/w clients + # print(f"client={client}") + response = await client.acompletion( + model="fake-openai-endpoint", # [CHANGE THIS] (if you call it something else on your proxy) + messages=[{"role": "user", "content": f"This is a test: {uuid.uuid4()}"}], + ) + return response + except Exception as e: + # print(e) + return None + +async def loadtest_fn(): + start = time.time() + n = 500 # Number of concurrent tasks + tasks = [router_completion_non_streaming() for _ in range(n)] + chat_completions = await asyncio.gather(*tasks) + successful_completions = [c for c in chat_completions if c is not None] + print(n, time.time() - start, len(successful_completions)) + +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 + + +# Run the event loop to execute the async function +async def parent_fn(): + for _ in range(10): + dt = get_utc_datetime() + current_minute = dt.strftime("%H-%M") + print(f"triggered new batch - {current_minute}") + await loadtest_fn() + await asyncio.sleep(10) + +asyncio.run(parent_fn()) + +``` + +### 2. Test it! + +Let's hit the router with 600 requests per minute. + +Copy this script 👇. Save it as `test_loadtest_router.py` AND run it with `python3 test_loadtest_router.py` + + +```python +from litellm import Router +import litellm +litellm.suppress_debug_info = True +litellm.set_verbose = False +import logging +logging.basicConfig(level=logging.CRITICAL) +import os, random, uuid, time, asyncio + +# Model list for OpenAI and Anthropic models +model_list = [ + { + "model_name": "fake-openai-endpoint", + "litellm_params": { + "model": "gpt-3.5-turbo", + "api_key": "my-fake-key", + "api_base": "http://0.0.0.0:8080", + "rpm": 100 + }, + }, + { + "model_name": "fake-openai-endpoint", + "litellm_params": { + "model": "gpt-3.5-turbo", + "api_key": "my-fake-key", + "api_base": "http://0.0.0.0:8081", + "rpm": 100 + }, + }, +] + +router_1 = Router(model_list=model_list, num_retries=0, enable_pre_call_checks=True, redis_host=os.getenv("REDIS_HOST"), redis_port=os.getenv("REDIS_PORT"), redis_password=os.getenv("REDIS_PASSWORD")) +router_2 = Router(model_list=model_list, num_retries=0, enable_pre_call_checks=True, redis_host=os.getenv("REDIS_HOST"), redis_port=os.getenv("REDIS_PORT"), redis_password=os.getenv("REDIS_PASSWORD")) + + + +async def router_completion_non_streaming(): + try: + client: Router = random.sample([router_1, router_2], 1)[0] # randomly pick b/w clients + # print(f"client={client}") + response = await client.acompletion( + model="fake-openai-endpoint", # [CHANGE THIS] (if you call it something else on your proxy) + messages=[{"role": "user", "content": f"This is a test: {uuid.uuid4()}"}], + ) + return response + except Exception as e: + # print(e) + return None + +async def loadtest_fn(): + start = time.time() + n = 600 # Number of concurrent tasks + tasks = [router_completion_non_streaming() for _ in range(n)] + chat_completions = await asyncio.gather(*tasks) + successful_completions = [c for c in chat_completions if c is not None] + print(n, time.time() - start, len(successful_completions)) + +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 + + +# Run the event loop to execute the async function +async def parent_fn(): + for _ in range(10): + dt = get_utc_datetime() + current_minute = dt.strftime("%H-%M") + print(f"triggered new batch - {current_minute}") + await loadtest_fn() + await asyncio.sleep(10) + +asyncio.run(parent_fn()) +``` +## Multi-Instance TPM/RPM Load Test (Proxy) Test if your defined tpm/rpm limits are respected across multiple instances. @@ -277,6 +461,8 @@ litellm --config /path/to/config.yaml --port 4001 Let's hit the proxy with 600 requests per minute. +Copy this script 👇. Save it as `test_loadtest_proxy.py` AND run it with `python3 test_loadtest_proxy.py` + ```python from openai import AsyncOpenAI, AsyncAzureOpenAI import random, uuid @@ -308,7 +494,7 @@ async def proxy_completion_non_streaming(): async def loadtest_fn(): start = time.time() - n = 500 # Number of concurrent tasks + n = 600 # Number of concurrent tasks tasks = [proxy_completion_non_streaming() for _ in range(n)] chat_completions = await asyncio.gather(*tasks) successful_completions = [c for c in chat_completions if c is not None]