From 1f5752dac9b561c3c996cd0aae539ff1fee05251 Mon Sep 17 00:00:00 2001 From: Ishaan Jaffer Date: Fri, 27 Feb 2026 18:21:49 -0800 Subject: [PATCH] perf: skip streaming hook overhead when no callbacks registered On every streaming chunk, async_data_generator was unconditionally: 1. Joining str_so_far_parts on every chunk (O(n^2) over the response) 2. Awaiting async_post_call_streaming_hook (with get_response_string parse) 3. Wrapping response in an extra async-generator layer All three run even when litellm.callbacks is empty, which is the common case for deployments without custom guardrails/loggers. Guard each path so the fast path is zero-cost when no callbacks registered. Benchmark (1000 concurrent, 167 chunks/response, no callbacks): before: 250.5 req/s | after: 305.3 req/s (+22% Python throughput) --- ...odel_prices_and_context_window_backup.json | 55 +++++++ litellm/proxy/proxy_server.py | 19 ++- litellm/proxy/utils.py | 143 +++++++++++------- ...st_post_call_streaming_hook_integration.py | 92 ++++++++++- 4 files changed, 241 insertions(+), 68 deletions(-) diff --git a/litellm/model_prices_and_context_window_backup.json b/litellm/model_prices_and_context_window_backup.json index 1ac8a347775..f52288ea72a 100644 --- a/litellm/model_prices_and_context_window_backup.json +++ b/litellm/model_prices_and_context_window_backup.json @@ -25157,6 +25157,25 @@ "supports_vision": true, "tool_use_system_prompt_tokens": 159 }, + "openrouter/anthropic/claude-opus-4.6": { + "cache_creation_input_token_cost": 6.25e-06, + "cache_read_input_token_cost": 5e-07, + "input_cost_per_token": 5e-06, + "litellm_provider": "openrouter", + "max_input_tokens": 1000000, + "max_output_tokens": 128000, + "max_tokens": 128000, + "mode": "chat", + "output_cost_per_token": 2.5e-05, + "supports_assistant_prefill": true, + "supports_computer_use": true, + "supports_function_calling": true, + "supports_prompt_caching": true, + "supports_reasoning": true, + "supports_tool_choice": true, + "supports_vision": true, + "tool_use_system_prompt_tokens": 346 + }, "openrouter/anthropic/claude-sonnet-4.5": { "input_cost_per_image": 0.0048, "cache_creation_input_token_cost": 3.75e-06, @@ -26169,6 +26188,42 @@ "supports_prompt_caching": true, "supports_computer_use": false }, + "openrouter/openrouter/auto": { + "input_cost_per_token": 0, + "output_cost_per_token": 0, + "litellm_provider": "openrouter", + "max_input_tokens": 2000000, + "max_tokens": 2000000, + "mode": "chat", + "supports_function_calling": true, + "supports_tool_choice": true, + "supports_reasoning": true, + "supports_response_schema": true, + "supports_vision": true, + "supports_audio_input": true, + "supports_video_input": true + }, + "openrouter/openrouter/free": { + "input_cost_per_token": 0, + "output_cost_per_token": 0, + "litellm_provider": "openrouter", + "max_input_tokens": 200000, + "max_tokens": 200000, + "mode": "chat", + "supports_function_calling": true, + "supports_tool_choice": true, + "supports_reasoning": true, + "supports_response_schema": true, + "supports_vision": true + }, + "openrouter/openrouter/bodybuilder": { + "input_cost_per_token": 0, + "output_cost_per_token": 0, + "litellm_provider": "openrouter", + "max_input_tokens": 128000, + "max_tokens": 128000, + "mode": "chat" + }, "ovhcloud/DeepSeek-R1-Distill-Llama-70B": { "input_cost_per_token": 6.7e-07, "litellm_provider": "ovhcloud", diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 48025863641..6796af09146 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -5315,14 +5315,19 @@ async def async_data_generator( request_data=request_data, ): ### CALL HOOKS ### - modify outgoing data - chunk = await proxy_logging_obj.async_post_call_streaming_hook( - user_api_key_dict=user_api_key_dict, - response=chunk, - data=request_data, - str_so_far="".join(str_so_far_parts), - ) + # Only compute str_so_far when callbacks are registered — joining + # the accumulated parts on every chunk is O(n²) otherwise. + if litellm.callbacks: + chunk = await proxy_logging_obj.async_post_call_streaming_hook( + user_api_key_dict=user_api_key_dict, + response=chunk, + data=request_data, + str_so_far="".join(str_so_far_parts), + ) - if isinstance(chunk, (ModelResponse, ModelResponseStream)): + if litellm.callbacks and isinstance( + chunk, (ModelResponse, ModelResponseStream) + ): response_str = litellm.get_response_string(response_obj=chunk) str_so_far_parts.append(response_str) diff --git a/litellm/proxy/utils.py b/litellm/proxy/utils.py index f6613b5548f..2c1a9beceff 100644 --- a/litellm/proxy/utils.py +++ b/litellm/proxy/utils.py @@ -23,23 +23,31 @@ from typing import ( ) from litellm import _custom_logger_compatible_callbacks_literal -from litellm.constants import (DEFAULT_MODEL_CREATED_AT_TIME, - MAX_TEAM_LIST_LIMIT) -from litellm.proxy._types import (DB_CONNECTION_ERROR_TYPES, CommonProxyErrors, - ProxyErrorTypes, ProxyException, - SpendLogsMetadata, SpendLogsPayload) +from litellm.constants import DEFAULT_MODEL_CREATED_AT_TIME, MAX_TEAM_LIST_LIMIT +from litellm.proxy._types import ( + DB_CONNECTION_ERROR_TYPES, + CommonProxyErrors, + ProxyErrorTypes, + ProxyException, + SpendLogsMetadata, + SpendLogsPayload, +) from litellm.types.guardrails import GuardrailEventHooks from litellm.types.utils import CallTypes, CallTypesLiteral try: - from litellm_enterprise.enterprise_callbacks.send_emails.base_email import \ - BaseEmailLogger - from litellm_enterprise.enterprise_callbacks.send_emails.resend_email import \ - ResendEmailLogger - from litellm_enterprise.enterprise_callbacks.send_emails.sendgrid_email import \ - SendGridEmailLogger - from litellm_enterprise.enterprise_callbacks.send_emails.smtp_email import \ - SMTPEmailLogger + from litellm_enterprise.enterprise_callbacks.send_emails.base_email import ( + BaseEmailLogger, + ) + from litellm_enterprise.enterprise_callbacks.send_emails.resend_email import ( + ResendEmailLogger, + ) + from litellm_enterprise.enterprise_callbacks.send_emails.sendgrid_email import ( + SendGridEmailLogger, + ) + from litellm_enterprise.enterprise_callbacks.send_emails.smtp_email import ( + SMTPEmailLogger, + ) except ImportError: BaseEmailLogger = None # type: ignore SendGridEmailLogger = None # type: ignore @@ -58,56 +66,70 @@ from fastapi import HTTPException, status import litellm import litellm.litellm_core_utils import litellm.litellm_core_utils.litellm_logging -from litellm import (EmbeddingResponse, ImageResponse, ModelResponse, - ModelResponseStream, Router) +from litellm import ( + EmbeddingResponse, + ImageResponse, + ModelResponse, + ModelResponseStream, + Router, +) from litellm._logging import verbose_proxy_logger from litellm._service_logger import ServiceLogging, ServiceTypes from litellm.caching.caching import DualCache, RedisCache from litellm.caching.dual_cache import LimitedSizeOrderedDict from litellm.exceptions import RejectedRequestError -from litellm.integrations.custom_guardrail import (CustomGuardrail, - ModifyResponseException) +from litellm.integrations.custom_guardrail import ( + CustomGuardrail, + ModifyResponseException, +) from litellm.integrations.custom_logger import CustomLogger from litellm.integrations.SlackAlerting.slack_alerting import SlackAlerting -from litellm.integrations.SlackAlerting.utils import \ - _add_langfuse_trace_id_to_alert +from litellm.integrations.SlackAlerting.utils import _add_langfuse_trace_id_to_alert from litellm.litellm_core_utils.litellm_logging import Logging from litellm.litellm_core_utils.safe_json_dumps import safe_dumps from litellm.litellm_core_utils.safe_json_loads import safe_json_loads from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler -from litellm.proxy._types import (AlertType, CallInfo, - LiteLLM_VerificationTokenView, Member, - UserAPIKeyAuth) +from litellm.proxy._types import ( + AlertType, + CallInfo, + LiteLLM_VerificationTokenView, + Member, + UserAPIKeyAuth, +) from litellm.proxy.auth.route_checks import RouteChecks -from litellm.proxy.db.create_views import (create_missing_views, - should_create_missing_views) +from litellm.proxy.db.create_views import ( + create_missing_views, + should_create_missing_views, +) from litellm.proxy.db.db_spend_update_writer import DBSpendUpdateWriter from litellm.proxy.db.exception_handler import PrismaDBExceptionHandler from litellm.proxy.db.log_db_metrics import log_db_metrics from litellm.proxy.db.prisma_client import PrismaWrapper -from litellm.proxy.guardrails.guardrail_hooks.unified_guardrail.unified_guardrail import \ - UnifiedLLMGuardrails +from litellm.proxy.guardrails.guardrail_hooks.unified_guardrail.unified_guardrail import ( + UnifiedLLMGuardrails, +) from litellm.proxy.hooks import PROXY_HOOKS, get_proxy_hook from litellm.proxy.hooks.cache_control_check import _PROXY_CacheControlCheck from litellm.proxy.hooks.max_budget_limiter import _PROXY_MaxBudgetLimiter -from litellm.proxy.hooks.parallel_request_limiter import \ - _PROXY_MaxParallelRequestsHandler +from litellm.proxy.hooks.parallel_request_limiter import ( + _PROXY_MaxParallelRequestsHandler, +) from litellm.proxy.litellm_pre_call_utils import LiteLLMProxyRequestSetup from litellm.proxy.policy_engine.pipeline_executor import PipelineExecutor from litellm.secret_managers.main import str_to_bool from litellm.types.integrations.slack_alerting import DEFAULT_ALERT_TYPES -from litellm.types.mcp import (MCPDuringCallResponseObject, - MCPPreCallRequestObject, - MCPPreCallResponseObject) -from litellm.types.proxy.policy_engine.pipeline_types import \ - PipelineExecutionResult +from litellm.types.mcp import ( + MCPDuringCallResponseObject, + MCPPreCallRequestObject, + MCPPreCallResponseObject, +) +from litellm.types.proxy.policy_engine.pipeline_types import PipelineExecutionResult from litellm.types.utils import LLMResponseTypes, LoggedLiteLLMParams if TYPE_CHECKING: from opentelemetry.trace import Span as _Span - from litellm.litellm_core_utils.litellm_logging import \ - Logging as LiteLLMLoggingObj + from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj Span = Union[_Span, Any] else: @@ -1050,9 +1072,10 @@ class ProxyLogging: """Process prompt template if applicable.""" from litellm.proxy.prompts.prompt_endpoints import ( - construct_versioned_prompt_id, get_latest_version_prompt_id) - from litellm.proxy.prompts.prompt_registry import \ - IN_MEMORY_PROMPT_REGISTRY + construct_versioned_prompt_id, + get_latest_version_prompt_id, + ) + from litellm.proxy.prompts.prompt_registry import IN_MEMORY_PROMPT_REGISTRY from litellm.utils import get_non_default_completion_params if prompt_version is None: @@ -1102,8 +1125,9 @@ class ProxyLogging: def _process_guardrail_metadata(self, data: dict) -> None: """Process guardrails from metadata and add to applied_guardrails.""" - from litellm.proxy.common_utils.callback_utils import \ - add_guardrail_to_applied_guardrails_header + from litellm.proxy.common_utils.callback_utils import ( + add_guardrail_to_applied_guardrails_header, + ) metadata_standard = data.get("metadata") or {} metadata_litellm = data.get("litellm_metadata") or {} @@ -1994,14 +2018,17 @@ class ProxyLogging: Covers: 1. /chat/completions """ + # Fast path: skip all per-chunk work when no callbacks are registered. + if not litellm.callbacks: + return response + from litellm.proxy.proxy_server import llm_router response_str: Optional[str] = None if isinstance(response, (ModelResponse, ModelResponseStream)): response_str = litellm.get_response_string(response_obj=response) elif isinstance(response, dict) and self.is_a2a_streaming_response(response): - from litellm.llms.a2a.common_utils import \ - extract_text_from_a2a_response + from litellm.llms.a2a.common_utils import extract_text_from_a2a_response response_str = extract_text_from_a2a_response(response) if response_str is not None: @@ -2010,8 +2037,7 @@ class ProxyLogging: _callback: Optional[CustomLogger] = None if isinstance(callback, CustomGuardrail): # Main - V2 Guardrails implementation - from litellm.types.guardrails import \ - GuardrailEventHooks + from litellm.types.guardrails import GuardrailEventHooks ## CHECK FOR MODEL-LEVEL GUARDRAILS modified_data = _check_and_merge_model_level_guardrails( @@ -2062,6 +2088,13 @@ class ProxyLogging: Covers: 1. /chat/completions """ + # Fast path: no callbacks registered — yield from the original iterator + # directly to avoid the extra async-generator wrapper overhead per chunk. + if not litellm.callbacks: + async for chunk in response: + yield chunk + return + current_response = response for callback in litellm.callbacks: @@ -4626,8 +4659,9 @@ async def update_spend_logs_job( # Guardrail/policy usage tracking (same batch, outside spend-logs update) try: - from litellm.proxy.guardrails.usage_tracking import \ - process_spend_logs_guardrail_usage + from litellm.proxy.guardrails.usage_tracking import ( + process_spend_logs_guardrail_usage, + ) await process_spend_logs_guardrail_usage( prisma_client=prisma_client, logs_to_process=logs_to_process, @@ -4653,8 +4687,10 @@ async def _monitor_spend_logs_queue( db_writer_client: Optional HTTP handler for external spend logs endpoint proxy_logging_obj: Proxy logging object """ - from litellm.constants import (SPEND_LOG_QUEUE_POLL_INTERVAL, - SPEND_LOG_QUEUE_SIZE_THRESHOLD) + from litellm.constants import ( + SPEND_LOG_QUEUE_POLL_INTERVAL, + SPEND_LOG_QUEUE_SIZE_THRESHOLD, + ) threshold = SPEND_LOG_QUEUE_SIZE_THRESHOLD base_interval = SPEND_LOG_QUEUE_POLL_INTERVAL @@ -5175,11 +5211,12 @@ async def get_available_models_for_user( List of model names available to the user """ from litellm.proxy.auth.auth_checks import get_team_object - from litellm.proxy.auth.model_checks import (get_complete_model_list, - get_key_models, - get_team_models) - from litellm.proxy.management_endpoints.team_endpoints import \ - validate_membership + from litellm.proxy.auth.model_checks import ( + get_complete_model_list, + get_key_models, + get_team_models, + ) + from litellm.proxy.management_endpoints.team_endpoints import validate_membership # Get proxy model list and access groups if llm_router is None: diff --git a/tests/test_litellm/proxy/hooks/test_post_call_streaming_hook_integration.py b/tests/test_litellm/proxy/hooks/test_post_call_streaming_hook_integration.py index 3bc111ef142..d9e882072cd 100644 --- a/tests/test_litellm/proxy/hooks/test_post_call_streaming_hook_integration.py +++ b/tests/test_litellm/proxy/hooks/test_post_call_streaming_hook_integration.py @@ -6,16 +6,17 @@ Tests verify that the streaming hook can transform streaming responses sent to c import os import sys -import pytest from typing import Any -from unittest.mock import patch, MagicMock +from unittest.mock import MagicMock, patch + +import pytest sys.path.insert(0, os.path.abspath("../../../..")) import litellm from litellm.integrations.custom_logger import CustomLogger from litellm.proxy._types import UserAPIKeyAuth -from litellm.types.utils import ModelResponseStream, StreamingChoices, Delta +from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices class StreamingResponseTransformerLogger(CustomLogger): @@ -46,8 +47,8 @@ async def test_streaming_hook_transforms_response(): transformer = StreamingResponseTransformerLogger(transform_content="Modified streaming response") with patch("litellm.callbacks", [transformer]): - from litellm.proxy.utils import ProxyLogging from litellm.caching.caching import DualCache + from litellm.proxy.utils import ProxyLogging proxy_logging = ProxyLogging(user_api_key_cache=DualCache()) @@ -101,8 +102,8 @@ async def test_streaming_hook_returns_none_keeps_original(): logger = NoOpLogger() with patch("litellm.callbacks", [logger]): - from litellm.proxy.utils import ProxyLogging from litellm.caching.caching import DualCache + from litellm.proxy.utils import ProxyLogging proxy_logging = ProxyLogging(user_api_key_cache=DualCache()) @@ -142,8 +143,8 @@ async def test_streaming_hook_works_with_sse_format(): ) with patch("litellm.callbacks", [transformer]): - from litellm.proxy.utils import ProxyLogging from litellm.caching.caching import DualCache + from litellm.proxy.utils import ProxyLogging proxy_logging = ProxyLogging(user_api_key_cache=DualCache()) @@ -195,8 +196,8 @@ async def test_streaming_hook_chains_multiple_callbacks(): callback2 = AppendLogger("CB2") with patch("litellm.callbacks", [callback1, callback2]): - from litellm.proxy.utils import ProxyLogging from litellm.caching.caching import DualCache + from litellm.proxy.utils import ProxyLogging proxy_logging = ProxyLogging(user_api_key_cache=DualCache()) @@ -228,6 +229,81 @@ async def test_streaming_hook_chains_multiple_callbacks(): assert result == "[CB2]" +@pytest.mark.asyncio +async def test_streaming_hook_fast_path_no_callbacks(): + """ + Test that async_post_call_streaming_hook returns immediately when no callbacks + are registered, without doing any per-chunk work. + """ + from litellm.caching.caching import DualCache + from litellm.proxy.utils import ProxyLogging + + with patch("litellm.callbacks", []): + proxy_logging = ProxyLogging(user_api_key_cache=DualCache()) + + original_response = ModelResponseStream( + id="fast-path-stream", + choices=[ + StreamingChoices( + delta=Delta(content="Hello", role="assistant"), + index=0, + ) + ], + model="test-model", + ) + + data = {"model": "test-model"} + user_api_key_dict = UserAPIKeyAuth(api_key="test-key") + + result = await proxy_logging.async_post_call_streaming_hook( + data=data, + response=original_response, + user_api_key_dict=user_api_key_dict, + ) + + # Should return the original response unchanged + assert result is original_response + + +@pytest.mark.asyncio +async def test_streaming_iterator_hook_fast_path_no_callbacks(): + """ + Test that async_post_call_streaming_iterator_hook yields chunks directly + without extra wrapping when no callbacks are registered. + """ + from litellm.caching.caching import DualCache + from litellm.proxy.utils import ProxyLogging + + chunks = [ + ModelResponseStream( + id=f"chunk-{i}", + choices=[StreamingChoices(delta=Delta(content=f"token{i}"), index=0)], + model="test-model", + ) + for i in range(5) + ] + + async def mock_response_iter(): + for chunk in chunks: + yield chunk + + with patch("litellm.callbacks", []): + proxy_logging = ProxyLogging(user_api_key_cache=DualCache()) + user_api_key_dict = UserAPIKeyAuth(api_key="test-key") + + collected = [] + async for chunk in proxy_logging.async_post_call_streaming_iterator_hook( + response=mock_response_iter(), + user_api_key_dict=user_api_key_dict, + request_data={}, + ): + collected.append(chunk) + + assert len(collected) == 5 + for i, chunk in enumerate(collected): + assert chunk.id == f"chunk-{i}" + + @pytest.mark.asyncio async def test_streaming_hook_handles_exceptions(): """ @@ -245,8 +321,8 @@ async def test_streaming_hook_handles_exceptions(): logger = FailingLogger() with patch("litellm.callbacks", [logger]): - from litellm.proxy.utils import ProxyLogging from litellm.caching.caching import DualCache + from litellm.proxy.utils import ProxyLogging proxy_logging = ProxyLogging(user_api_key_cache=DualCache())