From 1c6be9bdadbeac3ae66bac096e281bc305cd6b5b Mon Sep 17 00:00:00 2001 From: Perling Date: Tue, 5 Aug 2025 07:12:51 +0800 Subject: [PATCH] Fix: Langfuse reporting "client closed" error due to httpx client TTL (#13045) * Fix: Langfuse reporting "client closed" error due to httpx client TTL * remove log --- .../integrations/langfuse/langfuse_handler.py | 48 +++++++++++++++++-- 1 file changed, 45 insertions(+), 3 deletions(-) diff --git a/litellm/integrations/langfuse/langfuse_handler.py b/litellm/integrations/langfuse/langfuse_handler.py index f9d27f6cf00..56ed01563be 100644 --- a/litellm/integrations/langfuse/langfuse_handler.py +++ b/litellm/integrations/langfuse/langfuse_handler.py @@ -12,6 +12,10 @@ from litellm.litellm_core_utils.litellm_logging import StandardCallbackDynamicPa from .langfuse import LangFuseLogger, LangfuseLoggingConfig +from httpx._client import ClientState + +from litellm._logging import verbose_logger + if TYPE_CHECKING: from litellm.litellm_core_utils.litellm_logging import DynamicLoggingCache else: @@ -60,7 +64,17 @@ class LangFuseHandler: ) # if not cached, create a new langfuse logger and cache it - if temp_langfuse_logger is None: + if temp_langfuse_logger is None or not LangFuseHandler._logger_httpx_client_is_unclosed(temp_langfuse_logger): + # if the cached logger is closed, remove it from cache + if temp_langfuse_logger is not None: + key = in_memory_dynamic_logger_cache.get_cache_key( + args={**credentials_dict, "service_name": "langfuse"} + ) + in_memory_dynamic_logger_cache.cache._remove_key(key) + verbose_logger.warning( + "LangFuseLogger was found in cache but it was closed. Removing it from cache and creating a new one." + ) + temp_langfuse_logger = ( LangFuseHandler._create_langfuse_logger_from_credentials( credentials=credentials_dict, @@ -83,7 +97,7 @@ class LangFuseHandler: If no Global LangfuseLogger is set, it will check in_memory_dynamic_logger_cache for a cached LangFuseLogger This function is used to return the globalLangfuseLogger if it exists, otherwise it will check in_memory_dynamic_logger_cache for a cached LangFuseLogger """ - if globalLangfuseLogger is not None: + if globalLangfuseLogger is not None and LangFuseHandler._logger_httpx_client_is_unclosed(globalLangfuseLogger): return globalLangfuseLogger credentials_dict: Dict[ @@ -95,7 +109,17 @@ class LangFuseHandler: credentials=credentials_dict, service_name="langfuse", ) - if globalLangfuseLogger is None: + if globalLangfuseLogger is None or not LangFuseHandler._logger_httpx_client_is_unclosed(globalLangfuseLogger): + # if the cached logger is closed, remove it from cache + if globalLangfuseLogger is not None: + key = in_memory_dynamic_logger_cache.get_cache_key( + args={**credentials_dict, "service_name": "langfuse"} + ) + in_memory_dynamic_logger_cache.cache._remove_key(key) + verbose_logger.warning( + "LangFuseLogger was found in cache but it was closed. Removing it from cache and creating a new one." + ) + globalLangfuseLogger = ( LangFuseHandler._create_langfuse_logger_from_credentials( credentials=credentials_dict, @@ -168,3 +192,21 @@ class LangFuseHandler: ): return True return False + + @staticmethod + def _logger_httpx_client_is_unclosed( + logger: LangFuseLogger, + ) -> bool: + """ + This function checks if the httpx client used by the logger is not closed. + + Args: + logger (LangFuseLogger): The LangFuseLogger instance to check. + + Returns: + bool: True if the httpx client is not closed, False otherwise. + """ + verbose_logger.debug(f"LangFuseLogger's httpx client state: {logger.Langfuse.httpx_client._state}") + if logger is not None and logger.Langfuse.httpx_client._state != ClientState.CLOSED: + return True + return False