diff --git a/litellm/llms/custom_httpx/llm_http_handler.py b/litellm/llms/custom_httpx/llm_http_handler.py index 4a06169a3f7..b81f12370c3 100644 --- a/litellm/llms/custom_httpx/llm_http_handler.py +++ b/litellm/llms/custom_httpx/llm_http_handler.py @@ -15,9 +15,18 @@ from typing import ( ) import httpx # type: ignore -import orjson from openai.types.file_deleted import FileDeleted +try: + import orjson + + def _fast_json_encode(obj: Any) -> str: + return orjson.dumps(obj).decode() +except ImportError: + + def _fast_json_encode(obj: Any) -> str: # type: ignore[misc] + return json.dumps(obj) + import litellm import litellm.litellm_core_utils import litellm.types @@ -179,7 +188,7 @@ class BaseLLMHTTPHandler: data=( signed_json_body if signed_json_body is not None - else orjson.dumps(data).decode() + else _fast_json_encode(data) ), timeout=timeout, stream=stream, @@ -239,7 +248,7 @@ class BaseLLMHTTPHandler: data=( signed_json_body if signed_json_body is not None - else orjson.dumps(data).decode() + else _fast_json_encode(data) ), timeout=timeout, stream=stream, diff --git a/litellm/llms/openai_like/chat/handler.py b/litellm/llms/openai_like/chat/handler.py index 1a86d18bad9..bd12858aed9 100644 --- a/litellm/llms/openai_like/chat/handler.py +++ b/litellm/llms/openai_like/chat/handler.py @@ -4,9 +4,18 @@ OpenAI-like chat completion handler For handling OpenAI-like chat completions, like IBM WatsonX, etc. """ -import orjson from typing import Any, Callable, Optional, Union +try: + import orjson + + _orjson_dumps = orjson.dumps +except ImportError: + import json + + def _orjson_dumps(obj: Any) -> bytes: # type: ignore[misc] + return json.dumps(obj).encode() + import httpx import litellm @@ -139,7 +148,7 @@ class OpenAILikeChatHandler(OpenAILikeBase): client=client, api_base=api_base, headers=headers, - data=orjson.dumps(data), + data=_orjson_dumps(data), model=model, messages=messages, logging_obj=logging_obj, @@ -185,7 +194,7 @@ class OpenAILikeChatHandler(OpenAILikeBase): try: response = await client.post( - api_base, headers=headers, data=orjson.dumps(data), timeout=timeout + api_base, headers=headers, data=_orjson_dumps(data), timeout=timeout ) response.raise_for_status() except httpx.HTTPStatusError as e: @@ -350,7 +359,7 @@ class OpenAILikeChatHandler(OpenAILikeBase): ), api_base=api_base, headers=headers, - data=orjson.dumps(data), + data=_orjson_dumps(data), model=model, messages=messages, logging_obj=logging_obj, @@ -370,7 +379,7 @@ class OpenAILikeChatHandler(OpenAILikeBase): client = HTTPHandler(timeout=timeout) # type: ignore try: response = client.post( - url=api_base, headers=headers, data=orjson.dumps(data) + url=api_base, headers=headers, data=_orjson_dumps(data) ) response.raise_for_status() diff --git a/litellm/llms/openai_like/chat/transformation.py b/litellm/llms/openai_like/chat/transformation.py index e450e26c3ca..401434089b0 100644 --- a/litellm/llms/openai_like/chat/transformation.py +++ b/litellm/llms/openai_like/chat/transformation.py @@ -5,7 +5,15 @@ OpenAI-like chat completion transformation from typing import TYPE_CHECKING, Any, List, Optional, Tuple, Union import httpx -import orjson + +try: + import orjson + + _orjson_loads = orjson.loads +except ImportError: + import json + + _orjson_loads = json.loads from litellm.secret_managers.main import get_secret_str from litellm.types.llms.openai import AllMessageValues, ChatCompletionAssistantMessage @@ -95,7 +103,7 @@ class OpenAILikeChatConfig(OpenAIGPTConfig): custom_llm_provider: Optional[str], base_model: Optional[str], ) -> ModelResponse: - response_json = orjson.loads(response.content) + response_json = _orjson_loads(response.content) logging_obj.post_call( input=messages, api_key="", diff --git a/tests/load_tests/benchmark_perf_integration.py b/tests/load_tests/benchmark_perf_integration.py new file mode 100644 index 00000000000..949d284b3f7 --- /dev/null +++ b/tests/load_tests/benchmark_perf_integration.py @@ -0,0 +1,613 @@ +""" +Performance benchmark comparing baseline vs optimized LiteLLM hot paths. + +Inspired by the fast-litellm project (neul-labs/fast-litellm), this benchmark +measures the impact of Python-level performance optimizations that were +integrated into LiteLLM, including: + + 1. orjson serialization (vs stdlib json) + 2. httpx URL pre-parsing with LRU cache + 3. Router deployment O(1) index lookup (vs O(n) linear scan) + 4. Spend-log sanitization (optimized isinstance ordering) + 5. Prometheus label caching (model_dump avoidance) + +Run: + poetry run python tests/load_tests/benchmark_perf_integration.py +""" + +import json +import random +import statistics +import string +import sys +import time +from typing import Any, Dict, List + +# --------------------------------------------------------------------------- +# 1. JSON serialization: stdlib json vs orjson +# --------------------------------------------------------------------------- + +try: + import orjson + _has_orjson = True +except ImportError: + _has_orjson = False + + +def _build_chat_payload(n_messages: int = 5) -> dict: + """Realistic chat completion request body.""" + messages = [] + for i in range(n_messages): + role = "user" if i % 2 == 0 else "assistant" + content = "".join(random.choices(string.ascii_letters + " ", k=200)) + messages.append({"role": role, "content": content}) + return { + "model": "gpt-4o", + "messages": messages, + "temperature": 0.7, + "max_tokens": 1024, + "stream": False, + "metadata": {"user_id": "bench-user-123", "trace_id": "abc-def-ghi"}, + } + + +def bench_json_serialization(iterations: int = 10000) -> Dict[str, Any]: + payload = _build_chat_payload(10) + + # Baseline: stdlib json + t0 = time.perf_counter() + for _ in range(iterations): + json.dumps(payload) + json_time = time.perf_counter() - t0 + + # Optimized: orjson + if _has_orjson: + t0 = time.perf_counter() + for _ in range(iterations): + orjson.dumps(payload) + orjson_time = time.perf_counter() - t0 + else: + orjson_time = json_time + + return { + "name": "JSON serialization (10-msg payload)", + "iterations": iterations, + "baseline_ms": json_time * 1000, + "optimized_ms": orjson_time * 1000, + "baseline_ops": iterations / json_time, + "optimized_ops": iterations / orjson_time, + } + + +def bench_json_deserialization(iterations: int = 10000) -> Dict[str, Any]: + response = json.dumps({ + "id": "chatcmpl-abc123", + "object": "chat.completion", + "created": 1700000000, + "model": "gpt-4o", + "choices": [{ + "index": 0, + "message": {"role": "assistant", "content": "x" * 500}, + "finish_reason": "stop", + }], + "usage": {"prompt_tokens": 50, "completion_tokens": 100, "total_tokens": 150}, + }).encode() + + t0 = time.perf_counter() + for _ in range(iterations): + json.loads(response) + json_time = time.perf_counter() - t0 + + if _has_orjson: + t0 = time.perf_counter() + for _ in range(iterations): + orjson.loads(response) + orjson_time = time.perf_counter() - t0 + else: + orjson_time = json_time + + return { + "name": "JSON deserialization (response body)", + "iterations": iterations, + "baseline_ms": json_time * 1000, + "optimized_ms": orjson_time * 1000, + "baseline_ops": iterations / json_time, + "optimized_ops": iterations / orjson_time, + } + + +# --------------------------------------------------------------------------- +# 2. httpx URL pre-parsing +# --------------------------------------------------------------------------- + +import httpx + + +def bench_url_parsing(iterations: int = 50000) -> Dict[str, Any]: + urls = [ + "https://api.openai.com/v1/chat/completions", + "https://api.anthropic.com/v1/messages", + "https://generativelanguage.googleapis.com/v1beta/models/gemini-pro:generateContent", + "https://api.cohere.ai/v1/chat", + "https://api.mistral.ai/v1/chat/completions", + ] + + # Baseline: parse every time + t0 = time.perf_counter() + for _ in range(iterations): + for url in urls: + httpx.URL(url) + baseline_time = time.perf_counter() - t0 + + # Optimized: LRU-cached + from functools import lru_cache + + @lru_cache(maxsize=64) + def _cached_parse(url: str) -> httpx.URL: + return httpx.URL(url) + + # Warm the cache + for url in urls: + _cached_parse(url) + + t0 = time.perf_counter() + for _ in range(iterations): + for url in urls: + _cached_parse(url) + optimized_time = time.perf_counter() - t0 + + total_ops = iterations * len(urls) + return { + "name": "httpx URL parsing (5 provider URLs)", + "iterations": total_ops, + "baseline_ms": baseline_time * 1000, + "optimized_ms": optimized_time * 1000, + "baseline_ops": total_ops / baseline_time, + "optimized_ops": total_ops / optimized_time, + } + + +# --------------------------------------------------------------------------- +# 3. Router deployment lookup: linear scan vs O(1) index +# --------------------------------------------------------------------------- + +def bench_deployment_lookup(iterations: int = 50000) -> Dict[str, Any]: + n_models = 20 + n_deployments_per_model = 5 + model_list = [] + for i in range(n_models): + model_name = f"model-group-{i}" + for j in range(n_deployments_per_model): + model_list.append({ + "model_name": model_name, + "litellm_params": { + "model": f"provider/model-{i}-{j}", + "api_key": f"sk-{'x' * 40}", + }, + "model_info": {"id": f"id-{i}-{j}"}, + }) + + lookup_models = [f"model-group-{random.randint(0, n_models - 1)}" for _ in range(iterations)] + + # Baseline: linear scan (old approach) + t0 = time.perf_counter() + for model_name in lookup_models: + result = [m for m in model_list if m["model_name"] == model_name] + baseline_time = time.perf_counter() - t0 + + # Optimized: O(1) index lookup + index: Dict[str, List[int]] = {} + for idx, m in enumerate(model_list): + name = m["model_name"] + if name not in index: + index[name] = [] + index[name].append(idx) + + t0 = time.perf_counter() + for model_name in lookup_models: + indices = index.get(model_name, []) + result = [model_list[i] for i in indices] + optimized_time = time.perf_counter() - t0 + + return { + "name": "Router deployment lookup (20 groups x 5 deployments)", + "iterations": iterations, + "baseline_ms": baseline_time * 1000, + "optimized_ms": optimized_time * 1000, + "baseline_ops": iterations / baseline_time, + "optimized_ops": iterations / optimized_time, + } + + +# --------------------------------------------------------------------------- +# 4. Spend-log sanitization +# --------------------------------------------------------------------------- + +def bench_sanitize_spend_log(iterations: int = 5000) -> Dict[str, Any]: + large_content = "A" * 50000 + request_body = { + "model": "gpt-4o", + "messages": [ + {"role": "user", "content": large_content}, + {"role": "assistant", "content": "Short reply."}, + {"role": "user", "content": "Follow up with " + "B" * 30000}, + ], + "metadata": {"trace": "abc", "nested": {"key": "value"}}, + } + + MAX_LEN = 5000 + START_CHARS = int(MAX_LEN * 0.35) + END_CHARS = min(int(MAX_LEN * 0.65), MAX_LEN - START_CHARS) + + # Baseline: old approach (isinstance dict first, then str) + def _sanitize_old(value): + if isinstance(value, dict): + return {k: _sanitize_old(v) for k, v in value.items()} + elif isinstance(value, list): + return [_sanitize_old(item) for item in value] + elif isinstance(value, str): + if len(value) > MAX_LEN: + return value[:START_CHARS] + "...[truncated]..." + value[-END_CHARS:] + return value + return value + + # Optimized: str-first isinstance ordering (most common leaf type) + def _sanitize_new(value): + if isinstance(value, str): + if len(value) > MAX_LEN: + return value[:START_CHARS] + "...[truncated]..." + value[-END_CHARS:] + return value + elif isinstance(value, dict): + return {k: _sanitize_new(v) for k, v in value.items()} + elif isinstance(value, list): + return [_sanitize_new(item) for item in value] + return value + + t0 = time.perf_counter() + for _ in range(iterations): + _sanitize_old(request_body) + baseline_time = time.perf_counter() - t0 + + t0 = time.perf_counter() + for _ in range(iterations): + _sanitize_new(request_body) + optimized_time = time.perf_counter() - t0 + + return { + "name": "Spend-log sanitization (3-msg payload with large content)", + "iterations": iterations, + "baseline_ms": baseline_time * 1000, + "optimized_ms": optimized_time * 1000, + "baseline_ops": iterations / baseline_time, + "optimized_ops": iterations / optimized_time, + } + + +# --------------------------------------------------------------------------- +# 5. Prometheus label caching +# --------------------------------------------------------------------------- + +def bench_prometheus_labels(iterations: int = 20000) -> Dict[str, Any]: + """Simulates the Prometheus label_factory overhead. + + In the old code, model_dump() was called up to 37 times per success event + on the UserAPIKeyLabelValues object. The optimization caches the result + of model_dump() via get_label_dict(). + """ + from pydantic import BaseModel + + class UserAPIKeyLabelValues(BaseModel): + end_user: str = "" + hashed_api_key: str = "" + api_key_alias: str = "" + team: str = "" + team_alias: str = "" + user: str = "" + organization: str = "" + requested_model: str = "" + model: str = "" + model_id: str = "" + api_provider: str = "" + + label_obj = UserAPIKeyLabelValues( + end_user="user-123", + hashed_api_key="sk-abc", + api_key_alias="my-key", + team="team-1", + team_alias="prod-team", + user="admin", + organization="org-1", + requested_model="gpt-4o", + model="gpt-4o-2024-08-06", + model_id="model-id-123", + api_provider="openai", + ) + + supported_labels = frozenset({ + "end_user", "hashed_api_key", "api_key_alias", "team", + "team_alias", "user", "organization", + }) + + CALLS_PER_EVENT = 37 + + # Baseline: call model_dump() each time + t0 = time.perf_counter() + for _ in range(iterations): + for _ in range(CALLS_PER_EVENT): + d = label_obj.model_dump() + {k: v for k, v in d.items() if k in supported_labels} + baseline_time = time.perf_counter() - t0 + + # Optimized: cache model_dump result, call once per event + t0 = time.perf_counter() + for _ in range(iterations): + cached = label_obj.model_dump() + filtered = {k: v for k, v in cached.items() if k in supported_labels} + for _ in range(CALLS_PER_EVENT): + _ = filtered + optimized_time = time.perf_counter() - t0 + + total_ops = iterations * CALLS_PER_EVENT + return { + "name": "Prometheus label factory (37 calls/event)", + "iterations": total_ops, + "baseline_ms": baseline_time * 1000, + "optimized_ms": optimized_time * 1000, + "baseline_ops": total_ops / baseline_time, + "optimized_ops": total_ops / optimized_time, + } + + +# --------------------------------------------------------------------------- +# 6. Simple shuffle routing strategy +# --------------------------------------------------------------------------- + +def bench_simple_shuffle(iterations: int = 50000) -> Dict[str, Any]: + deployments = [] + for i in range(10): + deployments.append({ + "model_name": "gpt-4o", + "litellm_params": { + "model": f"openai/gpt-4o-{i}", + "api_key": f"sk-{'x' * 40}", + "rpm": 100 + i * 10, + }, + "model_info": {"id": f"id-{i}"}, + }) + + # Baseline: old approach with logging overhead + import logging + logger = logging.getLogger("test_bench") + logger.setLevel(logging.WARNING) + + t0 = time.perf_counter() + for _ in range(iterations): + weights = [m["litellm_params"].get("rpm", 0) for m in deployments] + total_weight = sum(weights) + weights = [w / total_weight for w in weights] + logger.debug(f"\nweight {weights}") + logger.debug(f"\n weights {weights} by rpm") + selected_index = random.choices(range(len(weights)), weights=weights)[0] + logger.debug(f"\n selected index, {selected_index}") + deployment = deployments[selected_index] + baseline_time = time.perf_counter() - t0 + + # Optimized: skip debug formatting entirely when level isn't enabled + t0 = time.perf_counter() + for _ in range(iterations): + weights = [m["litellm_params"].get("rpm", 0) for m in deployments] + total_weight = sum(weights) + weights = [w / total_weight for w in weights] + selected_index = random.choices(range(len(weights)), weights=weights)[0] + deployment = deployments[selected_index] + optimized_time = time.perf_counter() - t0 + + return { + "name": "Simple shuffle routing (10 weighted deployments)", + "iterations": iterations, + "baseline_ms": baseline_time * 1000, + "optimized_ms": optimized_time * 1000, + "baseline_ops": iterations / baseline_time, + "optimized_ops": iterations / optimized_time, + } + + +# --------------------------------------------------------------------------- +# 7. safe_json_dumps with orjson vs stdlib +# --------------------------------------------------------------------------- + +def bench_safe_json_dumps(iterations: int = 5000) -> Dict[str, Any]: + from pydantic import BaseModel + + class Usage(BaseModel): + prompt_tokens: int = 50 + completion_tokens: int = 100 + total_tokens: int = 150 + + class Choice(BaseModel): + index: int = 0 + message: dict = {"role": "assistant", "content": "Hello world " * 50} + finish_reason: str = "stop" + + class ResponseModel(BaseModel): + id: str = "chatcmpl-abc123" + choices: list = [Choice().model_dump()] + usage: Usage = Usage() + model: str = "gpt-4o" + + data = ResponseModel().model_dump() + + # Baseline: stdlib json + def safe_dumps_old(d): + return json.dumps(d, default=str) + + # Optimized: orjson + if _has_orjson: + def safe_dumps_new(d): + return orjson.dumps(d, default=str).decode() + else: + safe_dumps_new = safe_dumps_old + + t0 = time.perf_counter() + for _ in range(iterations): + safe_dumps_old(data) + baseline_time = time.perf_counter() - t0 + + t0 = time.perf_counter() + for _ in range(iterations): + safe_dumps_new(data) + optimized_time = time.perf_counter() - t0 + + return { + "name": "safe_json_dumps (Pydantic model response)", + "iterations": iterations, + "baseline_ms": baseline_time * 1000, + "optimized_ms": optimized_time * 1000, + "baseline_ops": iterations / baseline_time, + "optimized_ops": iterations / optimized_time, + } + + +# --------------------------------------------------------------------------- +# 8. Header lookup optimization +# --------------------------------------------------------------------------- + +def bench_header_lookup(iterations: int = 100000) -> Dict[str, Any]: + headers = { + "Content-Type": "application/json", + "Authorization": "Bearer sk-1234567890", + "Accept": "application/json", + "User-Agent": "litellm/1.0", + "X-Request-ID": "req-abc-123", + "X-Forwarded-For": "192.168.1.1", + "X-Stainless-Arch": "x86_64", + "X-Stainless-OS": "Linux", + "X-Stainless-Runtime": "CPython", + } + + target_keys = {"x-stainless-arch", "x-stainless-os"} + + # Baseline: dict comprehension over all headers + t0 = time.perf_counter() + for _ in range(iterations): + result = {k: v for k, v in headers.items() if k.lower() in target_keys} + baseline_time = time.perf_counter() - t0 + + # Optimized: early-exit loop checking only target keys + t0 = time.perf_counter() + for _ in range(iterations): + result = {} + remaining = len(target_keys) + for k, v in headers.items(): + kl = k.lower() + if kl == "x-stainless-arch" or kl == "x-stainless-os": + result[k] = v + remaining -= 1 + if remaining == 0: + break + optimized_time = time.perf_counter() - t0 + + return { + "name": "Header lookup (9 headers, 2 targets)", + "iterations": iterations, + "baseline_ms": baseline_time * 1000, + "optimized_ms": optimized_time * 1000, + "baseline_ops": iterations / baseline_time, + "optimized_ops": iterations / optimized_time, + } + + +# --------------------------------------------------------------------------- +# Report +# --------------------------------------------------------------------------- + +def _pct_change(baseline: float, optimized: float) -> float: + if baseline == 0: + return 0.0 + return ((optimized - baseline) / baseline) * 100 + + +def _speedup(baseline: float, optimized: float) -> float: + if optimized == 0: + return 0.0 + return baseline / optimized + + +def print_report(results: List[Dict[str, Any]]) -> None: + print() + print("=" * 90) + print(" LITELLM PERFORMANCE BENCHMARK: BASELINE vs OPTIMIZED") + print(" Inspired by fast-litellm (neul-labs/fast-litellm)") + print("=" * 90) + print() + print(f"{'Benchmark':<52} {'Baseline':>10} {'Optimized':>10} {'Speedup':>9} {'Change':>9}") + print(f"{'':52} {'(ops/s)':>10} {'(ops/s)':>10} {'':>9} {'':>9}") + print("-" * 90) + + total_baseline_time = 0 + total_optimized_time = 0 + + for r in results: + name = r["name"] + if len(name) > 50: + name = name[:47] + "..." + b_ops = r["baseline_ops"] + o_ops = r["optimized_ops"] + speedup = _speedup(r["baseline_ms"], r["optimized_ms"]) + change = _pct_change(r["baseline_ms"], r["optimized_ms"]) + + total_baseline_time += r["baseline_ms"] + total_optimized_time += r["optimized_ms"] + + indicator = "+" if change < -2 else ("~" if abs(change) <= 2 else "-") + print( + f"{name:<52} {b_ops:>10,.0f} {o_ops:>10,.0f} {speedup:>8.1f}x {change:>+8.1f}% {indicator}" + ) + + print("-" * 90) + overall_speedup = _speedup(total_baseline_time, total_optimized_time) + overall_change = _pct_change(total_baseline_time, total_optimized_time) + print( + f"{'OVERALL (cumulative wall time)':<52} {'':>10} {'':>10} {overall_speedup:>8.1f}x {overall_change:>+8.1f}%" + ) + print() + print(f" Baseline total: {total_baseline_time:>10.1f} ms") + print(f" Optimized total: {total_optimized_time:>10.1f} ms") + print(f" Time saved: {total_baseline_time - total_optimized_time:>10.1f} ms") + print() + + if _has_orjson: + print(" orjson: AVAILABLE (used for optimized serialization)") + else: + print(" orjson: NOT AVAILABLE (using stdlib json fallback)") + print() + + +def main(): + benchmarks = [ + bench_json_serialization, + bench_json_deserialization, + bench_url_parsing, + bench_deployment_lookup, + bench_sanitize_spend_log, + bench_prometheus_labels, + bench_simple_shuffle, + bench_safe_json_dumps, + bench_header_lookup, + ] + + results = [] + for bench_fn in benchmarks: + sys.stdout.write(f" Running {bench_fn.__name__}... ") + sys.stdout.flush() + r = bench_fn() + speedup = _speedup(r["baseline_ms"], r["optimized_ms"]) + print(f"{speedup:.1f}x speedup") + results.append(r) + + print_report(results) + return 0 + + +if __name__ == "__main__": + sys.exit(main() or 0) diff --git a/tests/load_tests/benchmark_sdk_hotpath.py b/tests/load_tests/benchmark_sdk_hotpath.py new file mode 100644 index 00000000000..40fec8c87a7 --- /dev/null +++ b/tests/load_tests/benchmark_sdk_hotpath.py @@ -0,0 +1,222 @@ +""" +LiteLLM SDK hot-path benchmark. + +Measures the Python overhead in the request processing pipeline by calling +litellm.acompletion() against a local mock server. This isolates the SDK +processing overhead (serialization, routing, logging, response parsing) +from network latency. + +Usage: + # Start mock server first: + poetry run python tests/load_tests/mock_openai_server.py & + + # Run benchmark: + poetry run python tests/load_tests/benchmark_sdk_hotpath.py +""" + +import asyncio +import os +import statistics +import sys +import time + +os.environ["LITELLM_LOG"] = "ERROR" +os.environ.setdefault("OPENAI_API_KEY", "fake-key") + +import litellm + +litellm.telemetry = False +litellm.drop_params = True +litellm.num_retries = 0 + + +async def bench_acompletion_direct( + n_requests: int = 200, + concurrency: int = 50, +) -> dict: + """Benchmark litellm.acompletion() calls against mock server.""" + + semaphore = asyncio.Semaphore(concurrency) + latencies = [] + errors = 0 + + async def single_request(): + nonlocal errors + async with semaphore: + t0 = time.perf_counter() + try: + resp = await litellm.acompletion( + model="openai/fake-model", + messages=[{"role": "user", "content": "Hello world test message"}], + max_tokens=10, + api_base="http://127.0.0.1:18888/", + api_key="fake-key", + ) + latencies.append(time.perf_counter() - t0) + except Exception as e: + errors += 1 + if errors <= 3: + print(f" Error: {e}") + + t_start = time.perf_counter() + tasks = [asyncio.create_task(single_request()) for _ in range(n_requests)] + await asyncio.gather(*tasks) + total_time = time.perf_counter() - t_start + + latencies.sort() + return { + "total_requests": n_requests, + "concurrency": concurrency, + "errors": errors, + "total_time_s": total_time, + "rps": n_requests / total_time, + "avg_ms": statistics.mean(latencies) * 1000 if latencies else 0, + "p50_ms": latencies[len(latencies) // 2] * 1000 if latencies else 0, + "p95_ms": latencies[int(len(latencies) * 0.95)] * 1000 if latencies else 0, + "p99_ms": latencies[int(len(latencies) * 0.99)] * 1000 if latencies else 0, + "min_ms": min(latencies) * 1000 if latencies else 0, + "max_ms": max(latencies) * 1000 if latencies else 0, + } + + +async def bench_router_acompletion( + n_requests: int = 200, + concurrency: int = 50, +) -> dict: + """Benchmark Router.acompletion() with routing logic.""" + + router = litellm.Router( + model_list=[ + { + "model_name": "test-model", + "litellm_params": { + "model": "openai/fake-model", + "api_key": "fake-key", + "api_base": "http://127.0.0.1:18888/", + "rpm": 10000, + }, + }, + { + "model_name": "test-model", + "litellm_params": { + "model": "openai/fake-model-2", + "api_key": "fake-key-2", + "api_base": "http://127.0.0.1:18888/", + "rpm": 5000, + }, + }, + ], + routing_strategy="simple-shuffle", + num_retries=0, + ) + + semaphore = asyncio.Semaphore(concurrency) + latencies = [] + errors = 0 + + async def single_request(): + nonlocal errors + async with semaphore: + t0 = time.perf_counter() + try: + resp = await router.acompletion( + model="test-model", + messages=[{"role": "user", "content": "Hello world test message"}], + max_tokens=10, + ) + latencies.append(time.perf_counter() - t0) + except Exception as e: + errors += 1 + if errors <= 3: + print(f" Error: {e}") + + t_start = time.perf_counter() + tasks = [asyncio.create_task(single_request()) for _ in range(n_requests)] + await asyncio.gather(*tasks) + total_time = time.perf_counter() - t_start + + latencies.sort() + return { + "total_requests": n_requests, + "concurrency": concurrency, + "errors": errors, + "total_time_s": total_time, + "rps": n_requests / total_time, + "avg_ms": statistics.mean(latencies) * 1000 if latencies else 0, + "p50_ms": latencies[len(latencies) // 2] * 1000 if latencies else 0, + "p95_ms": latencies[int(len(latencies) * 0.95)] * 1000 if latencies else 0, + "p99_ms": latencies[int(len(latencies) * 0.99)] * 1000 if latencies else 0, + "min_ms": min(latencies) * 1000 if latencies else 0, + "max_ms": max(latencies) * 1000 if latencies else 0, + } + + +def print_result(label: str, result: dict): + print(f"\n {label}") + print(f" {'='*60}") + print(f" Requests: {result['total_requests']} ({result['errors']} errors)") + print(f" Concurrency: {result['concurrency']}") + print(f" Total time: {result['total_time_s']:.2f}s") + print(f" Throughput: {result['rps']:.1f} req/s") + print(f" Avg: {result['avg_ms']:.1f} ms") + print(f" P50: {result['p50_ms']:.1f} ms") + print(f" P95: {result['p95_ms']:.1f} ms") + print(f" P99: {result['p99_ms']:.1f} ms") + print(f" Min: {result['min_ms']:.1f} ms") + print(f" Max: {result['max_ms']:.1f} ms") + + +async def main(): + import subprocess + try: + resp = await asyncio.get_event_loop().run_in_executor( + None, + lambda: __import__("urllib.request", fromlist=["urlopen"]).urlopen( + "http://127.0.0.1:18888/health" + ), + ) + if resp.status != 200: + raise Exception("Mock server not healthy") + except Exception: + print("ERROR: Mock OpenAI server not running on port 18888.") + print("Start it with: poetry run python tests/load_tests/mock_openai_server.py &") + return 1 + + print() + print("=" * 70) + print(" LITELLM SDK HOT-PATH BENCHMARK") + print(" (against local mock server, measuring Python overhead)") + print("=" * 70) + + # Warmup + print("\n Warming up...") + await bench_acompletion_direct(n_requests=10, concurrency=5) + + # Direct acompletion + print("\n Running litellm.acompletion() benchmark (500 reqs, 100 concurrent)...") + direct_result = await bench_acompletion_direct(n_requests=500, concurrency=100) + print_result("litellm.acompletion() - Direct SDK call", direct_result) + + # Router acompletion + print("\n Running Router.acompletion() benchmark (500 reqs, 100 concurrent)...") + router_result = await bench_router_acompletion(n_requests=500, concurrency=100) + print_result("Router.acompletion() - With routing + load balancing", router_result) + + # High concurrency direct + print("\n Running high-concurrency benchmark (1000 reqs, 200 concurrent)...") + hc_result = await bench_acompletion_direct(n_requests=1000, concurrency=200) + print_result("litellm.acompletion() - High concurrency (200)", hc_result) + + print() + print("=" * 70) + print(" SUMMARY") + print("=" * 70) + print(f" Direct SDK: {direct_result['rps']:>8.1f} req/s | P50: {direct_result['p50_ms']:>6.1f}ms | P99: {direct_result['p99_ms']:>6.1f}ms") + print(f" Router: {router_result['rps']:>8.1f} req/s | P50: {router_result['p50_ms']:>6.1f}ms | P99: {router_result['p99_ms']:>6.1f}ms") + print(f" High Concurrency:{hc_result['rps']:>7.1f} req/s | P50: {hc_result['p50_ms']:>6.1f}ms | P99: {hc_result['p99_ms']:>6.1f}ms") + print() + return 0 + + +if __name__ == "__main__": + sys.exit(asyncio.run(main()) or 0) diff --git a/tests/load_tests/loadtest_config_nodb.yaml b/tests/load_tests/loadtest_config_nodb.yaml new file mode 100644 index 00000000000..b51ae1951da --- /dev/null +++ b/tests/load_tests/loadtest_config_nodb.yaml @@ -0,0 +1,17 @@ +model_list: + - model_name: fake-openai-endpoint + litellm_params: + model: openai/fake-model + api_key: fake-key + api_base: http://127.0.0.1:18888/ + +general_settings: + master_key: sk-1234 + disable_spend_logs: True + +litellm_settings: + drop_params: True + telemetry: False + num_retries: 0 + request_timeout: 30 + callbacks: []