diff --git a/docs/my-website/docs/proxy/config_settings.md b/docs/my-website/docs/proxy/config_settings.md index fdd68c953f6..f8d219c6a02 100644 --- a/docs/my-website/docs/proxy/config_settings.md +++ b/docs/my-website/docs/proxy/config_settings.md @@ -528,6 +528,7 @@ router_settings: | MAX_TOKEN_TRIMMING_ATTEMPTS | Maximum number of attempts to trim a token message. Default is 10 | MAXIMUM_TRACEBACK_LINES_TO_LOG | Maximum number of lines to log in traceback in LiteLLM Logs UI. Default is 100 | MAX_RETRY_DELAY | Maximum delay in seconds for retrying requests. Default is 8.0 +| MAX_LANGFUSE_INITIALIZED_CLIENTS | Maximum number of Langfuse clients to initialize on proxy. Default is 20. This is set since langfuse initializes 1 thread everytime a client is initialized. We've had an incident in the past where we reached 100% cpu utilization because Langfuse was initialized several times. | MIN_NON_ZERO_TEMPERATURE | Minimum non-zero temperature value. Default is 0.0001 | MINIMUM_PROMPT_CACHE_TOKEN_COUNT | Minimum token count for caching a prompt. Default is 1024 | MISTRAL_API_BASE | Base URL for Mistral API diff --git a/litellm/__init__.py b/litellm/__init__.py index 5061c8e7e5b..5e40d3e3d8d 100644 --- a/litellm/__init__.py +++ b/litellm/__init__.py @@ -126,6 +126,7 @@ _known_custom_logger_compatible_callbacks: List = list( callbacks: List[ Union[Callable, _custom_logger_compatible_callbacks_literal, CustomLogger] ] = [] +initialized_langfuse_clients: int = 0 langfuse_default_tags: Optional[List[str]] = None langsmith_batch_size: Optional[int] = None prometheus_initialize_budget_metrics: Optional[bool] = False diff --git a/litellm/constants.py b/litellm/constants.py index 148cb9847c8..bb5d56978be 100644 --- a/litellm/constants.py +++ b/litellm/constants.py @@ -153,6 +153,9 @@ FIREWORKS_AI_16_B = int(os.getenv("FIREWORKS_AI_16_B", 16)) FIREWORKS_AI_80_B = int(os.getenv("FIREWORKS_AI_80_B", 80)) #### Logging callback constants #### REDACTED_BY_LITELM_STRING = "REDACTED_BY_LITELM" +MAX_LANGFUSE_INITIALIZED_CLIENTS = int( + os.getenv("MAX_LANGFUSE_INITIALIZED_CLIENTS", 20) +) ############### LLM Provider Constants ############### ### ANTHROPIC CONSTANTS ### diff --git a/litellm/integrations/SlackAlerting/utils.py b/litellm/integrations/SlackAlerting/utils.py index 0dc8bae5a6a..e695266c88b 100644 --- a/litellm/integrations/SlackAlerting/utils.py +++ b/litellm/integrations/SlackAlerting/utils.py @@ -5,6 +5,7 @@ Utils used for slack alerting import asyncio from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union +import litellm from litellm.proxy._types import AlertType from litellm.secret_managers.main import get_secret @@ -69,7 +70,12 @@ async def _add_langfuse_trace_id_to_alert( -> trace_id -> litellm_call_id """ - # do nothing for now + if "langfuse" not in litellm.logging_callback_manager._get_all_callbacks(): + return None + ######################################################### + # Only run if langfuse is added as a callback + ######################################################### + if ( request_data is not None and request_data.get("litellm_logging_obj", None) is not None @@ -82,11 +88,12 @@ async def _add_langfuse_trace_id_to_alert( if trace_id is not None: break await asyncio.sleep(3) # wait 3s before retrying for trace id - - _langfuse_object = litellm_logging_obj._get_callback_object( + ######################################################### + langfuse_object = litellm_logging_obj._get_callback_object( service_name="langfuse" ) - if _langfuse_object is not None: - base_url = _langfuse_object.Langfuse.base_url + if langfuse_object is not None: + base_url = langfuse_object.Langfuse.base_url return f"{base_url}/trace/{trace_id}" + return None diff --git a/litellm/integrations/langfuse/langfuse.py b/litellm/integrations/langfuse/langfuse.py index 02862c52c5d..2674f2ace0a 100644 --- a/litellm/integrations/langfuse/langfuse.py +++ b/litellm/integrations/langfuse/langfuse.py @@ -10,6 +10,7 @@ from packaging.version import Version import litellm from litellm._logging import verbose_logger +from litellm.constants import MAX_LANGFUSE_INITIALIZED_CLIENTS from litellm.litellm_core_utils.redact_messages import redact_user_api_key_info from litellm.llms.custom_httpx.http_handler import _get_httpx_client from litellm.secret_managers.main import str_to_bool @@ -27,12 +28,13 @@ from litellm.types.utils import ( ) if TYPE_CHECKING: - from langfuse.client import StatefulTraceClient + from langfuse.client import Langfuse, StatefulTraceClient from litellm.litellm_core_utils.litellm_logging import DynamicLoggingCache else: DynamicLoggingCache = Any StatefulTraceClient = Any + Langfuse = Any class LangFuseLogger: @@ -84,8 +86,7 @@ class LangFuseLogger: if Version(self.langfuse_sdk_version) >= Version("2.6.0"): parameters["sdk_integration"] = "litellm" - - self.Langfuse = Langfuse(**parameters) + self.Langfuse: Langfuse = self.safe_init_langfuse_client(parameters) # set the current langfuse project id in the environ # this is used by Alerting to link to the correct project @@ -124,6 +125,24 @@ class LangFuseLogger: else: self.upstream_langfuse = None + def safe_init_langfuse_client(self, parameters: dict) -> Langfuse: + """ + Safely init a langfuse client if the number of initialized clients is less than the max + + Note: + - Langfuse initializes 1 thread everytime a client is initialized. + - We've had an incident in the past where we reached 100% cpu utilization because Langfuse was initialized several times. + """ + from langfuse import Langfuse + + if litellm.initialized_langfuse_clients >= MAX_LANGFUSE_INITIALIZED_CLIENTS: + raise Exception( + f"Max langfuse clients reached: {litellm.initialized_langfuse_clients} is greater than {MAX_LANGFUSE_INITIALIZED_CLIENTS}" + ) + langfuse_client = Langfuse(**parameters) + litellm.initialized_langfuse_clients += 1 + return langfuse_client + @staticmethod def add_metadata_from_header(litellm_params: dict, metadata: dict) -> dict: """ diff --git a/litellm/litellm_core_utils/logging_callback_manager.py b/litellm/litellm_core_utils/logging_callback_manager.py index dec3add4e1b..e1bddc65497 100644 --- a/litellm/litellm_core_utils/logging_callback_manager.py +++ b/litellm/litellm_core_utils/logging_callback_manager.py @@ -266,3 +266,12 @@ class LoggingCallbackManager: if isinstance(callback, callback_type) and callback not in all_callbacks: all_callbacks.append(callback) return all_callbacks + + def callback_is_active(self, callback_type: Type[CustomLogger]) -> bool: + """ + Returns True if any of the active callbacks are of the given type + """ + return any( + isinstance(callback, callback_type) + for callback in self._get_all_callbacks() + ) diff --git a/litellm/proxy/proxy_config.yaml b/litellm/proxy/proxy_config.yaml index ae0c4b2de48..2e37fbe9776 100644 --- a/litellm/proxy/proxy_config.yaml +++ b/litellm/proxy/proxy_config.yaml @@ -2,25 +2,10 @@ model_list: - model_name: openai/gpt-4o litellm_params: model: openai/gpt-4o - api_key: any_key - api_base: https://exampleopenaiendpoint-production.up.railway.app/ + api_key: a + api_base: hi general_settings: store_prompts_in_spend_logs: true - - - -guardrails: - - guardrail_name: "custom-pre-guard" - litellm_params: - guardrail: custom_guardrail.myCustomGuardrail # 👈 Key change - mode: "pre_call" # runs async_pre_call_hook - - guardrail_name: "custom-during-guard" - litellm_params: - guardrail: custom_guardrail.myCustomGuardrail - mode: "during_call" # runs async_moderation_hook - - guardrail_name: "custom-post-guard" - litellm_params: - guardrail: custom_guardrail.myCustomGuardrail - mode: "post_call" # runs async_post_call_success_hook \ No newline at end of file + alerting: ["slack"] \ No newline at end of file diff --git a/tests/litellm/integrations/SlackAlerting/test_slack_alerting_utils.py b/tests/litellm/integrations/SlackAlerting/test_slack_alerting_utils.py new file mode 100644 index 00000000000..027fed1b5ff --- /dev/null +++ b/tests/litellm/integrations/SlackAlerting/test_slack_alerting_utils.py @@ -0,0 +1,39 @@ +import json +import os +import sys +from typing import Optional +from unittest.mock import MagicMock + +import pytest + +# Adds the grandparent directory to sys.path to allow importing project modules +sys.path.insert(0, os.path.abspath("../..")) + +import litellm +from litellm.integrations.langfuse.langfuse_prompt_management import ( + LangfusePromptManagement, +) +from litellm.integrations.SlackAlerting.utils import _add_langfuse_trace_id_to_alert +from litellm.litellm_core_utils.logging_callback_manager import LoggingCallbackManager + + +@pytest.mark.asyncio +async def test_langfuse_not_initialized_returns_none_early(): + """ + Test that when no LangfusePromptManagement is initialized, + the function returns None immediately without executing further logic + """ + # Ensure no Langfuse logger is in the callback manager + litellm.logging_callback_manager = LoggingCallbackManager() + + # Create request data that would normally trigger processing + request_data = {"litellm_logging_obj": MagicMock(), "trace_id": "test-trace-id"} + + # Call the function + result = await _add_langfuse_trace_id_to_alert(request_data) + + # Should return None early without processing request_data + assert result is None + + # Verify the litellm_logging_obj was never accessed (early return) + request_data["litellm_logging_obj"].assert_not_called() diff --git a/tests/litellm/integrations/test_langfuse.py b/tests/litellm/integrations/test_langfuse.py new file mode 100644 index 00000000000..7e47f76b995 --- /dev/null +++ b/tests/litellm/integrations/test_langfuse.py @@ -0,0 +1,57 @@ +import json +import os +import sys +from typing import Optional + +# Adds the grandparent directory to sys.path to allow importing project modules +sys.path.insert(0, os.path.abspath("../..")) + +import asyncio +from unittest.mock import patch + +import pytest + +import litellm +from litellm.integrations.langfuse.langfuse import LangFuseLogger + + +def test_max_langfuse_clients_limit(): + """ + Test that the max langfuse clients limit is respected when initializing multiple clients + """ + # Set max clients to 2 for testing + with patch( + "litellm.integrations.langfuse.langfuse.MAX_LANGFUSE_INITIALIZED_CLIENTS", 2 + ): + # Reset the counter + litellm.initialized_langfuse_clients = 0 + + # First client should succeed + logger1 = LangFuseLogger( + langfuse_public_key="test_key_1", + langfuse_secret="test_secret_1", + langfuse_host="https://test1.langfuse.com", + ) + assert litellm.initialized_langfuse_clients == 1 + + # Second client should succeed + logger2 = LangFuseLogger( + langfuse_public_key="test_key_2", + langfuse_secret="test_secret_2", + langfuse_host="https://test2.langfuse.com", + ) + assert litellm.initialized_langfuse_clients == 2 + + # Third client should fail with exception + with pytest.raises(Exception) as exc_info: + logger3 = LangFuseLogger( + langfuse_public_key="test_key_3", + langfuse_secret="test_secret_3", + langfuse_host="https://test3.langfuse.com", + ) + + # Verify the error message contains the expected text + assert "Max langfuse clients reached" in str(exc_info.value) + + # Counter should still be 2 (third client failed to initialize) + assert litellm.initialized_langfuse_clients == 2