[Fix] Reliability Fix - Removing code that was creating threads on errors (#11066)

* fix: only init langfuse if active

* fix: only init langfuse if active

* fix: add initialized_langfuse_clients count

* fix: add MAX_LANGFUSE_INITIALIZED_CLIENTS

* fix: use safe init langfuse

* test: init langfuse clients

* test: test_langfuse_not_initialized_returns_none_early

* docs MAX_LANGFUSE_INITIALIZED_CLIENTS

* fix: use correct langfuse callback

* fix: code qa

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

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

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 ###

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

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

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

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
+  alerting: ["slack"]

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

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