diff --git a/litellm-sidecar/src/main.rs b/litellm-sidecar/src/main.rs index 0cab82d8886..febf6ba668e 100644 --- a/litellm-sidecar/src/main.rs +++ b/litellm-sidecar/src/main.rs @@ -37,7 +37,7 @@ impl Sidecar { return client.clone(); } let client = Client::builder() - .pool_max_idle_per_host(100) + .pool_max_idle_per_host(200) .pool_idle_timeout(std::time::Duration::from_secs(90)) .tcp_keepalive(std::time::Duration::from_secs(60)) .tcp_nodelay(true) @@ -57,19 +57,14 @@ async fn handle_request( let path = req.uri().path(); if path == "/health" { + let total = sidecar.total_latency_us.load(Ordering::Relaxed); + let count = sidecar.total_requests.load(Ordering::Relaxed); + let avg = if count > 0 { total / count } else { 0 }; let stats = format!( r#"{{"status":"ok","requests":{},"errors":{},"avg_latency_us":{}}}"#, - sidecar.total_requests.load(Ordering::Relaxed), + count, sidecar.total_errors.load(Ordering::Relaxed), - { - let total = sidecar.total_latency_us.load(Ordering::Relaxed); - let count = sidecar.total_requests.load(Ordering::Relaxed); - if count > 0 { - total / count - } else { - 0 - } - } + avg, ); return Ok(Response::builder() .status(200) @@ -78,6 +73,7 @@ async fn handle_request( .unwrap()); } + // Extract routing metadata from X-LiteLLM-* headers let provider_url = match req.headers().get("x-litellm-provider-url") { Some(v) => v.to_str().unwrap_or("").to_string(), None => { @@ -85,16 +81,21 @@ async fn handle_request( return Ok(Response::builder() .status(400) .body( - Full::new(Bytes::from( - r#"{"error":"Missing X-LiteLLM-Provider-URL header"}"#, - )) - .map_err(|e| match e {}) - .boxed(), + Full::new(Bytes::from(r#"{"error":"Missing X-LiteLLM-Provider-URL"}"#)) + .map_err(|e| match e {}) + .boxed(), ) .unwrap()); } }; + let request_path = req + .headers() + .get("x-litellm-path") + .and_then(|v| v.to_str().ok()) + .unwrap_or("/chat/completions") + .to_string(); + let api_key = req .headers() .get("x-litellm-api-key") @@ -116,13 +117,24 @@ async fn handle_request( .map(|v| v == "true") .unwrap_or(false); - let request_path = req + let content_type = req .headers() - .get("x-litellm-path") + .get("content-type") .and_then(|v| v.to_str().ok()) - .unwrap_or("/chat/completions") + .unwrap_or("application/json") .to_string(); + // Collect forwarded headers (X-LiteLLM-Fwd-*) + let mut fwd_headers: Vec<(String, String)> = Vec::new(); + for (name, value) in req.headers() { + let name_str = name.as_str(); + if let Some(orig_name) = name_str.strip_prefix("x-litellm-fwd-") { + if let Ok(val) = value.to_str() { + fwd_headers.push((orig_name.to_string(), val.to_string())); + } + } + } + let body_bytes = match req.collect().await { Ok(collected) => collected.to_bytes(), Err(_) => { @@ -130,11 +142,9 @@ async fn handle_request( return Ok(Response::builder() .status(400) .body( - Full::new(Bytes::from( - r#"{"error":"Failed to read request body"}"#, - )) - .map_err(|e| match e {}) - .boxed(), + Full::new(Bytes::from(r#"{"error":"Failed to read request body"}"#)) + .map_err(|e| match e {}) + .boxed(), ) .unwrap()); } @@ -145,12 +155,11 @@ async fn handle_request( .unwrap_or_else(|_| "unknown".to_string()); let client = sidecar.get_or_create_client(&host); - let full_url = format!("{}{}", provider_url.trim_end_matches('/'), request_path); let mut req_builder = client .post(&full_url) - .header("content-type", "application/json") + .header("content-type", &content_type) .timeout(std::time::Duration::from_secs(timeout_secs)) .body(body_bytes.to_vec()); @@ -158,23 +167,21 @@ async fn handle_request( req_builder = req_builder.header("authorization", format!("Bearer {}", api_key)); } + for (name, value) in &fwd_headers { + req_builder = req_builder.header(name.as_str(), value.as_str()); + } + let provider_response = match req_builder.send().await { Ok(resp) => resp, Err(e) => { sidecar.total_errors.fetch_add(1, Ordering::Relaxed); let elapsed = start.elapsed().as_micros() as u64; - sidecar - .total_latency_us - .fetch_add(elapsed, Ordering::Relaxed); + sidecar.total_latency_us.fetch_add(elapsed, Ordering::Relaxed); let error_body = format!(r#"{{"error":"Provider request failed: {}"}}"#, e); return Ok(Response::builder() .status(502) .header("content-type", "application/json") - .body( - Full::new(Bytes::from(error_body)) - .map_err(|e| match e {}) - .boxed(), - ) + .body(Full::new(Bytes::from(error_body)).map_err(|e| match e {}).boxed()) .unwrap()); } }; @@ -182,23 +189,20 @@ async fn handle_request( let status = provider_response.status(); let status_code = StatusCode::from_u16(status.as_u16()).unwrap_or(StatusCode::BAD_GATEWAY); + // Forward response headers from provider + let mut resp_builder = Response::builder().status(status_code); + for (name, value) in provider_response.headers() { + resp_builder = resp_builder.header(name.as_str(), value.as_bytes()); + } + if is_stream { let stream = provider_response.bytes_stream().map(|result| { - let frame = result - .map(Frame::data) - .unwrap_or_else(|_| Frame::data(Bytes::new())); - Ok::<_, Infallible>(frame) + Ok::<_, Infallible>(Frame::data(result.unwrap_or_default())) }); - let body = StreamBody::new(stream); let elapsed = start.elapsed().as_micros() as u64; - sidecar - .total_latency_us - .fetch_add(elapsed, Ordering::Relaxed); - Ok(Response::builder() - .status(status_code) - .header("content-type", "text/event-stream") - .header("transfer-encoding", "chunked") - .body(BodyExt::boxed(body)) + sidecar.total_latency_us.fetch_add(elapsed, Ordering::Relaxed); + Ok(resp_builder + .body(BodyExt::boxed(StreamBody::new(stream))) .unwrap()) } else { let resp_bytes = match provider_response.bytes().await { @@ -206,33 +210,18 @@ async fn handle_request( Err(e) => { sidecar.total_errors.fetch_add(1, Ordering::Relaxed); let elapsed = start.elapsed().as_micros() as u64; - sidecar - .total_latency_us - .fetch_add(elapsed, Ordering::Relaxed); - let error_body = - format!(r#"{{"error":"Failed to read provider response: {}"}}"#, e); + sidecar.total_latency_us.fetch_add(elapsed, Ordering::Relaxed); + let error_body = format!(r#"{{"error":"Failed to read response: {}"}}"#, e); return Ok(Response::builder() .status(502) - .body( - Full::new(Bytes::from(error_body)) - .map_err(|e| match e {}) - .boxed(), - ) + .body(Full::new(Bytes::from(error_body)).map_err(|e| match e {}).boxed()) .unwrap()); } }; let elapsed = start.elapsed().as_micros() as u64; - sidecar - .total_latency_us - .fetch_add(elapsed, Ordering::Relaxed); - Ok(Response::builder() - .status(status_code) - .header("content-type", "application/json") - .body( - Full::new(resp_bytes) - .map_err(|e| match e {}) - .boxed(), - ) + sidecar.total_latency_us.fetch_add(elapsed, Ordering::Relaxed); + Ok(resp_builder + .body(Full::new(resp_bytes).map_err(|e| match e {}).boxed()) .unwrap()) } } diff --git a/litellm/llms/custom_httpx/http_handler.py b/litellm/llms/custom_httpx/http_handler.py index 3dfef07d426..7847e9f9d5c 100644 --- a/litellm/llms/custom_httpx/http_handler.py +++ b/litellm/llms/custom_httpx/http_handler.py @@ -737,18 +737,17 @@ class AsyncHTTPHandler: ) -> Optional[Union[LiteLLMAiohttpTransport, AsyncHTTPTransport]]: """ - Creates a transport for httpx.AsyncClient + - if USE_SIDECAR is set, use the Rust sidecar transport - if litellm.force_ipv4 is True, it will return AsyncHTTPTransport with local_address="0.0.0.0" - [Default] It will return AiohttpTransport - Users can opt out of using AiohttpTransport by setting litellm.use_aiohttp_transport to False - - - Notes on this handler: - - Why AiohttpTransport? - - By default, we use AiohttpTransport since it offers much higher throughput and lower latency than httpx. - - - Why force ipv4? - - Some users have seen httpx ConnectionError when using ipv6 - forcing ipv4 resolves the issue for them """ + ######################################################### + # SIDECAR TRANSPORT — Rust-based HTTP forwarding + ######################################################### + if AsyncHTTPHandler._should_use_sidecar_transport(): + return AsyncHTTPHandler._create_sidecar_transport() + ######################################################### # AIOHTTP TRANSPORT is off by default ######################################################### @@ -764,6 +763,23 @@ class AsyncHTTPHandler: ######################################################### return AsyncHTTPHandler._create_httpx_transport() + @staticmethod + def _should_use_sidecar_transport() -> bool: + """Check if the Rust sidecar transport is enabled via env var.""" + return os.getenv("USE_SIDECAR", "").lower() == "true" + + @staticmethod + def _create_sidecar_transport(): + """Create a sidecar transport that forwards requests through the Rust binary.""" + from litellm.llms.custom_httpx.sidecar_transport import ( + LiteLLMSidecarTransport, + ) + + port = int(os.getenv("SIDECAR_PORT", "8787")) + sidecar_url = f"http://127.0.0.1:{port}" + verbose_logger.info("Using Sidecar transport → %s", sidecar_url) + return LiteLLMSidecarTransport(sidecar_url=sidecar_url) + @staticmethod def _should_use_aiohttp_transport() -> bool: """ diff --git a/litellm/llms/custom_httpx/llm_http_handler.py b/litellm/llms/custom_httpx/llm_http_handler.py index 1cef3e9ce15..7eccc8fe71c 100644 --- a/litellm/llms/custom_httpx/llm_http_handler.py +++ b/litellm/llms/custom_httpx/llm_http_handler.py @@ -1,4 +1,5 @@ import json +import orjson import ssl from typing import ( TYPE_CHECKING, @@ -180,7 +181,7 @@ class BaseLLMHTTPHandler: data=( signed_json_body if signed_json_body is not None - else json.dumps(data) + else orjson.dumps(data) ), timeout=timeout, stream=stream, @@ -240,7 +241,7 @@ class BaseLLMHTTPHandler: data=( signed_json_body if signed_json_body is not None - else json.dumps(data) + else orjson.dumps(data) ), timeout=timeout, stream=stream, diff --git a/litellm/llms/custom_httpx/sidecar_transport.py b/litellm/llms/custom_httpx/sidecar_transport.py new file mode 100644 index 00000000000..c4dfe5fa51b --- /dev/null +++ b/litellm/llms/custom_httpx/sidecar_transport.py @@ -0,0 +1,119 @@ +""" +Rust sidecar transport for httpx. + +Drop-in replacement for LiteLLMAiohttpTransport that forwards HTTP requests +through the Rust sidecar binary. All LiteLLM provider transformations, +logging, callbacks, and retry logic still run in Python — only the TCP +connection pooling and HTTP round-trip move to Rust. +""" + +import typing +from typing import Optional + +import aiohttp +import httpx + + +class SidecarResponseStream(httpx.AsyncByteStream): + """Wraps an aiohttp response as an httpx async byte stream.""" + + CHUNK_SIZE = 1024 * 16 + + def __init__(self, response: aiohttp.ClientResponse) -> None: + self._response = response + + async def __aiter__(self) -> typing.AsyncIterator[bytes]: + async for chunk in self._response.content.iter_chunked(self.CHUNK_SIZE): + yield chunk + + async def aclose(self) -> None: + self._response.close() + + +class LiteLLMSidecarTransport(httpx.AsyncBaseTransport): + """ + httpx transport that forwards requests through the Rust sidecar. + + The sidecar handles connection pooling, timeout enforcement, and + metrics aggregation in Rust — eliminating GIL contention for the + HTTP I/O path under high concurrency. + """ + + def __init__(self, sidecar_url: str = "http://127.0.0.1:8787"): + self._sidecar_url = sidecar_url + self._session: Optional[aiohttp.ClientSession] = None + + def _get_session(self) -> aiohttp.ClientSession: + if self._session is None or self._session.closed: + self._session = aiohttp.ClientSession( + connector=aiohttp.TCPConnector( + limit=0, + keepalive_timeout=90, + enable_cleanup_closed=True, + ), + timeout=aiohttp.ClientTimeout(total=None, connect=5), + ) + return self._session + + async def handle_async_request(self, request: httpx.Request) -> httpx.Response: + # Extract the provider host from the full URL for the sidecar headers + parsed = request.url + provider_base = f"{parsed.scheme}://{parsed.host}" + if parsed.port and parsed.port not in (80, 443): + provider_base += f":{parsed.port}" + path = parsed.raw_path.decode("ascii") if isinstance(parsed.raw_path, bytes) else str(parsed.raw_path) + + # Extract auth header if present + api_key = "" + auth = request.headers.get("authorization", "") + if auth.lower().startswith("bearer "): + api_key = auth[7:] + + # Determine if streaming from content-type or accept headers + is_stream = "text/event-stream" in request.headers.get("accept", "") + + # Get timeout from request extensions + timeout_config = request.extensions.get("timeout", {}) + timeout_secs = timeout_config.get("read", 300) or 300 + if isinstance(timeout_secs, (int, float)): + timeout_secs = int(timeout_secs) + else: + timeout_secs = 300 + + try: + body = request.content + except httpx.RequestNotRead: + body = b"" + + headers = { + "X-LiteLLM-Provider-URL": provider_base, + "X-LiteLLM-API-Key": api_key, + "X-LiteLLM-Timeout": str(timeout_secs), + "X-LiteLLM-Stream": "true" if is_stream else "false", + "X-LiteLLM-Path": path, + "Content-Type": request.headers.get("content-type", "application/json"), + } + + # Copy provider-specific headers the sidecar should forward + for key in ("x-api-key", "anthropic-version", "x-goog-api-key"): + val = request.headers.get(key) + if val: + headers[f"X-LiteLLM-Fwd-{key}"] = val + + session = self._get_session() + resp = await session.post( + f"{self._sidecar_url}/forward", + data=body, + headers=headers, + ) + + return httpx.Response( + status_code=resp.status, + headers=dict(resp.headers), + stream=SidecarResponseStream(resp), + request=request, + ) + + async def aclose(self) -> None: + if self._session and not self._session.closed: + await self._session.close() diff --git a/litellm/llms/openai_like/chat/handler.py b/litellm/llms/openai_like/chat/handler.py index 821fc9b7f15..1a86d18bad9 100644 --- a/litellm/llms/openai_like/chat/handler.py +++ b/litellm/llms/openai_like/chat/handler.py @@ -4,7 +4,7 @@ OpenAI-like chat completion handler For handling OpenAI-like chat completions, like IBM WatsonX, etc. """ -import json +import orjson from typing import Any, Callable, Optional, Union import httpx @@ -139,7 +139,7 @@ class OpenAILikeChatHandler(OpenAILikeBase): client=client, api_base=api_base, headers=headers, - data=json.dumps(data), + data=orjson.dumps(data), model=model, messages=messages, logging_obj=logging_obj, @@ -185,7 +185,7 @@ class OpenAILikeChatHandler(OpenAILikeBase): try: response = await client.post( - api_base, headers=headers, data=json.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 +350,7 @@ class OpenAILikeChatHandler(OpenAILikeBase): ), api_base=api_base, headers=headers, - data=json.dumps(data), + data=orjson.dumps(data), model=model, messages=messages, logging_obj=logging_obj, @@ -370,7 +370,7 @@ class OpenAILikeChatHandler(OpenAILikeBase): client = HTTPHandler(timeout=timeout) # type: ignore try: response = client.post( - url=api_base, headers=headers, data=json.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 1c8cd574c01..e450e26c3ca 100644 --- a/litellm/llms/openai_like/chat/transformation.py +++ b/litellm/llms/openai_like/chat/transformation.py @@ -5,6 +5,7 @@ OpenAI-like chat completion transformation from typing import TYPE_CHECKING, Any, List, Optional, Tuple, Union import httpx +import orjson from litellm.secret_managers.main import get_secret_str from litellm.types.llms.openai import AllMessageValues, ChatCompletionAssistantMessage @@ -94,7 +95,7 @@ class OpenAILikeChatConfig(OpenAIGPTConfig): custom_llm_provider: Optional[str], base_model: Optional[str], ) -> ModelResponse: - response_json = response.json() + response_json = orjson.loads(response.content) logging_obj.post_call( input=messages, api_key="", diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index df52e354c5c..216459ab132 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -6639,21 +6639,25 @@ async def model_info( "/v1/chat/completions", dependencies=[Depends(user_api_key_auth)], tags=["chat/completions"], + response_class=ORJSONResponse, ) @router.post( "/chat/completions", dependencies=[Depends(user_api_key_auth)], tags=["chat/completions"], + response_class=ORJSONResponse, ) @router.post( "/engines/{model:path}/chat/completions", dependencies=[Depends(user_api_key_auth)], tags=["chat/completions"], + response_class=ORJSONResponse, ) @router.post( "/openai/deployments/{model:path}/chat/completions", dependencies=[Depends(user_api_key_auth)], tags=["chat/completions"], + response_class=ORJSONResponse, responses={200: {"description": "Successful response"}, **ERROR_RESPONSES}, ) # azure compatible endpoint async def chat_completion( # noqa: PLR0915 diff --git a/litellm/proxy/route_llm_request.py b/litellm/proxy/route_llm_request.py index 555f25652da..1b791980af3 100644 --- a/litellm/proxy/route_llm_request.py +++ b/litellm/proxy/route_llm_request.py @@ -143,38 +143,6 @@ def add_shared_session_to_data(data: dict) -> None: pass -async def _try_sidecar_route( - data: dict, - llm_router: Optional[LitellmRouter], - route_type: str, -) -> "Any | None": - """ - Attempt to route an acompletion request through the Rust sidecar. - Returns None if the sidecar is not available or not applicable. - """ - if route_type != "acompletion": - return None - - from litellm.proxy.sidecar_handler import is_sidecar_enabled, sidecar_acompletion - - if not is_sidecar_enabled(): - return None - if llm_router is None: - return None - - try: - deployment = await llm_router.async_get_available_deployment( - model=data.get("model", ""), - messages=data.get("messages"), - request_kwargs=data, - ) - if deployment is None: - return None - return await sidecar_acompletion(data, deployment) - except Exception: - return None - - async def route_request( # noqa: PLR0915 - Complex routing function, refactoring tracked separately data: dict, llm_router: Optional[LitellmRouter], @@ -253,14 +221,6 @@ async def route_request( # noqa: PLR0915 - Complex routing function, refactorin """ Common helper to route the request """ - # Fast path: try Rust sidecar for acompletion requests - sidecar_result = await _try_sidecar_route(data, llm_router, route_type) - if sidecar_result is not None: - # Wrap in a coroutine to match the expected interface (result goes into asyncio.gather) - async def _resolved(): - return sidecar_result - return _resolved() - add_shared_session_to_data(data) team_id = get_team_id_from_data(data) diff --git a/litellm/proxy/sidecar_handler.py b/litellm/proxy/sidecar_handler.py deleted file mode 100644 index 41fd89ae0bf..00000000000 --- a/litellm/proxy/sidecar_handler.py +++ /dev/null @@ -1,119 +0,0 @@ -""" -Sidecar request handler. - -Translates between LiteLLM's internal data format and the sidecar's -HTTP forwarding protocol. Returns ModelResponse objects compatible -with the rest of the proxy pipeline. -""" - -import json -from typing import Optional - -from litellm.proxy.sidecar_client import get_sidecar_client -from litellm.types.utils import ModelResponse - - -def _extract_provider_info(data: dict, deployment: Optional[dict] = None) -> dict: - """Extract provider URL, API key, and other forwarding metadata from request data.""" - info = { - "api_base": "", - "api_key": "", - "timeout": 300, - "stream": False, - "model": "", - } - - if deployment: - litellm_params = deployment.get("litellm_params", {}) - info["api_base"] = litellm_params.get("api_base", "") - info["api_key"] = litellm_params.get("api_key", "") - info["timeout"] = litellm_params.get("timeout", 300) - info["model"] = litellm_params.get("model", "") - - # Override with request-level data - if "api_base" in data: - info["api_base"] = data["api_base"] - if "api_key" in data: - info["api_key"] = data["api_key"] - if "timeout" in data: - info["timeout"] = data["timeout"] - if "stream" in data: - info["stream"] = data["stream"] - if "model" in data: - info["model"] = data["model"] - - return info - - -async def sidecar_acompletion(data: dict, deployment: dict) -> ModelResponse: - """ - Forward a chat completion request through the sidecar. - - Returns a ModelResponse compatible with litellm's response format. - """ - client = get_sidecar_client() - if client is None or not client.is_healthy: - raise RuntimeError("Sidecar not available") - - provider_info = _extract_provider_info(data, deployment) - - # Build the request body (what the provider expects) - request_body = { - "model": provider_info["model"].split("/", 1)[-1] - if "/" in provider_info["model"] - else provider_info["model"], - "messages": data.get("messages", []), - } - - # Forward optional params - for key in [ - "max_tokens", - "temperature", - "top_p", - "n", - "stop", - "presence_penalty", - "frequency_penalty", - "logit_bias", - "user", - "response_format", - "seed", - "tools", - "tool_choice", - "stream", - ]: - if key in data and data[key] is not None: - request_body[key] = data[key] - - timeout = provider_info["timeout"] - if isinstance(timeout, (int, float)): - timeout = int(timeout) - else: - timeout = 300 - - resp = await client.forward_request( - provider_url=provider_info["api_base"], - api_key=provider_info["api_key"], - request_body=request_body, - path="/v1/chat/completions", - timeout=timeout, - stream=False, - ) - - resp_body = await resp.read() - resp_json = json.loads(resp_body) - - if resp.status != 200: - raise Exception( - f"Sidecar forwarding failed with status {resp.status}: {resp_json}" - ) - - # Convert to ModelResponse - model_response = ModelResponse(**resp_json) - return model_response - - -def is_sidecar_enabled() -> bool: - """Check if the sidecar is enabled and healthy.""" - client = get_sidecar_client() - return client is not None and client.is_healthy