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)
This commit is contained in:
Ishaan Jaffer 2026-02-27 18:21:49 -08:00
parent 15fcd90b9c
commit 1f5752dac9
4 changed files with 241 additions and 68 deletions

View file

@ -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",

View file

@ -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)

View file

@ -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:

View file

@ -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())