perf: move async/sync callback separation from per-request to registration time

The three loops in function_setup that called is_async_callable() on every
callback each request were redundant after the first request. Move the
async/sync routing into LoggingCallbackManager.add_litellm_*_callback()
so it happens once at registration time instead of on every request.

litellm/litellm_core_utils/logging_callback_manager.py

@@ -27,13 +27,28 @@ class LoggingCallbackManager:
     # healthy maximum number of callbacks - unlikely someone needs more than 20
     MAX_CALLBACKS = 30
 
-    def add_litellm_input_callback(self, callback: Union[CustomLogger, str]):
+    def _is_async_callable(self, callback) -> bool:
+        """Check if a callback is async. Used to auto-route callbacks to the correct list."""
+        try:
+            from litellm.litellm_core_utils.coroutine_checker import coroutine_checker
+
+            return coroutine_checker.is_async_callable(callback)
+        except Exception:
+            return False
+
+    def add_litellm_input_callback(self, callback: Union[CustomLogger, str, Callable]):
         """
-        Add a input callback to litellm.input_callback
+        Add a input callback to litellm.input_callback.
+        Auto-routes async callbacks to litellm._async_input_callback.
         """
-        self._safe_add_callback_to_list(
-            callback=callback, parent_list=litellm.input_callback
-        )
+        if not isinstance(callback, str) and self._is_async_callable(callback):
+            self._safe_add_callback_to_list(
+                callback=callback, parent_list=litellm._async_input_callback
+            )
+        else:
+            self._safe_add_callback_to_list(
+                callback=callback, parent_list=litellm.input_callback
+            )
 
     def add_litellm_service_callback(
         self, callback: Union[CustomLogger, str, Callable]
@@ -59,21 +74,38 @@ class LoggingCallbackManager:
         self, callback: Union[CustomLogger, str, Callable]
     ):
         """
-        Add a success callback to `litellm.success_callback`
+        Add a success callback to `litellm.success_callback`.
+        Auto-routes async callbacks to litellm._async_success_callback.
+        Special-cases 'dynamodb' and 'openmeter' as async callbacks.
         """
-        self._safe_add_callback_to_list(
-            callback=callback, parent_list=litellm.success_callback
-        )
+        if isinstance(callback, str) and callback in ("dynamodb", "openmeter"):
+            self._safe_add_callback_to_list(
+                callback=callback, parent_list=litellm._async_success_callback
+            )
+        elif not isinstance(callback, str) and self._is_async_callable(callback):
+            self._safe_add_callback_to_list(
+                callback=callback, parent_list=litellm._async_success_callback
+            )
+        else:
+            self._safe_add_callback_to_list(
+                callback=callback, parent_list=litellm.success_callback
+            )
 
     def add_litellm_failure_callback(
         self, callback: Union[CustomLogger, str, Callable]
     ):
         """
-        Add a failure callback to `litellm.failure_callback`
+        Add a failure callback to `litellm.failure_callback`.
+        Auto-routes async callbacks to litellm._async_failure_callback.
         """
-        self._safe_add_callback_to_list(
-            callback=callback, parent_list=litellm.failure_callback
-        )
+        if not isinstance(callback, str) and self._is_async_callable(callback):
+            self._safe_add_callback_to_list(
+                callback=callback, parent_list=litellm._async_failure_callback
+            )
+        else:
+            self._safe_add_callback_to_list(
+                callback=callback, parent_list=litellm.failure_callback
+            )
 
     def add_litellm_async_success_callback(
         self, callback: Union[CustomLogger, Callable, str]

litellm/utils.py

@@ -826,59 +826,6 @@ def function_setup(  # noqa: PLR0915
             )
             get_set_callbacks = getattr(sys.modules[__name__], "get_set_callbacks")
             get_set_callbacks()(callback_list=callback_list, function_id=function_id)
-        ## ASYNC CALLBACKS
-        if len(litellm.input_callback) > 0:
-            removed_async_items = []
-            for index, callback in enumerate(litellm.input_callback):  # type: ignore
-                if coroutine_checker.is_async_callable(callback):
-                    litellm._async_input_callback.append(callback)
-                    removed_async_items.append(index)
-
-            # Pop the async items from input_callback in reverse order to avoid index issues
-            for index in reversed(removed_async_items):
-                litellm.input_callback.pop(index)
-        if len(litellm.success_callback) > 0:
-            removed_async_items = []
-            for index, callback in enumerate(litellm.success_callback):  # type: ignore
-                if coroutine_checker.is_async_callable(callback):
-                    litellm.logging_callback_manager.add_litellm_async_success_callback(
-                        callback
-                    )
-                    removed_async_items.append(index)
-                elif callback == "dynamodb" or callback == "openmeter":
-                    # dynamo is an async callback, it's used for the proxy and needs to be async
-                    # we only support async dynamo db logging for acompletion/aembedding since that's used on proxy
-                    litellm.logging_callback_manager.add_litellm_async_success_callback(
-                        callback
-                    )
-                    removed_async_items.append(index)
-                elif (
-                    callback in litellm._known_custom_logger_compatible_callbacks
-                    and isinstance(callback, str)
-                ):
-                    _add_custom_logger_callback_to_specific_event(callback, "success")
-
-            # Pop the async items from success_callback in reverse order to avoid index issues
-            for index in reversed(removed_async_items):
-                litellm.success_callback.pop(index)
-
-        if len(litellm.failure_callback) > 0:
-            removed_async_items = []
-            for index, callback in enumerate(litellm.failure_callback):  # type: ignore
-                if coroutine_checker.is_async_callable(callback):
-                    litellm.logging_callback_manager.add_litellm_async_failure_callback(
-                        callback
-                    )
-                    removed_async_items.append(index)
-                elif (
-                    callback in litellm._known_custom_logger_compatible_callbacks
-                    and isinstance(callback, str)
-                ):
-                    _add_custom_logger_callback_to_specific_event(callback, "failure")
-
-            # Pop the async items from failure_callback in reverse order to avoid index issues
-            for index in reversed(removed_async_items):
-                litellm.failure_callback.pop(index)
         ### DYNAMIC CALLBACKS ###
         dynamic_success_callbacks: Optional[
             List[Union[str, Callable, "CustomLogger"]]

tests/litellm_utils_tests/test_logging_callback_manager.py

@@ -269,11 +269,11 @@ async def test_slack_alerting_callback_registration(callback_manager):
             alert_types=["outage_alerts"]
         )
         assert len(litellm.callbacks) == 1  # Regular callback for outage alerts
-        assert len(litellm.success_callback) == 1  # Success callback for response_taking_too_long
         assert isinstance(litellm.callbacks[0], SlackAlerting)
-        # Get the method reference for comparison
+        # response_taking_too_long_callback is async, so it should be in the async success callback list
         response_taking_too_long_callback = proxy_logging.slack_alerting_instance.response_taking_too_long_callback
-        assert litellm.success_callback[0] == response_taking_too_long_callback
+        assert len(litellm._async_success_callback) == 1
+        assert litellm._async_success_callback[0] == response_taking_too_long_callback
 
         # Cleanup
         callback_manager._reset_all_callbacks()

tests/test_litellm/test_utils.py

@@ -3304,3 +3304,79 @@ class TestIsStreamingRequest:
 
     def test_stream_true_overrides_non_streaming_call_type(self):
         assert _is_streaming_request(kwargs={"stream": True}, call_type=CallTypes.acompletion) is True
+
+
+class TestCallbackAsyncSyncSeparation:
+    """Test that LoggingCallbackManager auto-routes async callbacks to async lists."""
+
+    def setup_method(self):
+        """Reset callback lists before each test."""
+        litellm.input_callback = []
+        litellm.success_callback = []
+        litellm.failure_callback = []
+        litellm._async_input_callback = []
+        litellm._async_success_callback = []
+        litellm._async_failure_callback = []
+
+    def test_async_success_callback_routed_to_async_list(self):
+        async def my_async_cb(*args, **kwargs):
+            pass
+
+        litellm.logging_callback_manager.add_litellm_success_callback(my_async_cb)
+        assert my_async_cb in litellm._async_success_callback
+        assert my_async_cb not in litellm.success_callback
+
+    def test_sync_success_callback_stays_in_sync_list(self):
+        def my_sync_cb(*args, **kwargs):
+            pass
+
+        litellm.logging_callback_manager.add_litellm_success_callback(my_sync_cb)
+        assert my_sync_cb in litellm.success_callback
+        assert my_sync_cb not in litellm._async_success_callback
+
+    def test_string_callback_stays_in_sync_list(self):
+        litellm.logging_callback_manager.add_litellm_success_callback("langfuse")
+        assert "langfuse" in litellm.success_callback
+        assert "langfuse" not in litellm._async_success_callback
+
+    def test_async_failure_callback_routed_to_async_list(self):
+        async def my_async_cb(*args, **kwargs):
+            pass
+
+        litellm.logging_callback_manager.add_litellm_failure_callback(my_async_cb)
+        assert my_async_cb in litellm._async_failure_callback
+        assert my_async_cb not in litellm.failure_callback
+
+    def test_sync_failure_callback_stays_in_sync_list(self):
+        def my_sync_cb(*args, **kwargs):
+            pass
+
+        litellm.logging_callback_manager.add_litellm_failure_callback(my_sync_cb)
+        assert my_sync_cb in litellm.failure_callback
+        assert my_sync_cb not in litellm._async_failure_callback
+
+    def test_dynamodb_routed_to_async_success(self):
+        litellm.logging_callback_manager.add_litellm_success_callback("dynamodb")
+        assert "dynamodb" in litellm._async_success_callback
+        assert "dynamodb" not in litellm.success_callback
+
+    def test_openmeter_routed_to_async_success(self):
+        litellm.logging_callback_manager.add_litellm_success_callback("openmeter")
+        assert "openmeter" in litellm._async_success_callback
+        assert "openmeter" not in litellm.success_callback
+
+    def test_async_input_callback_routed_to_async_list(self):
+        async def my_async_cb(*args, **kwargs):
+            pass
+
+        litellm.logging_callback_manager.add_litellm_input_callback(my_async_cb)
+        assert my_async_cb in litellm._async_input_callback
+        assert my_async_cb not in litellm.input_callback
+
+    def test_sync_input_callback_stays_in_sync_list(self):
+        def my_sync_cb(*args, **kwargs):
+            pass
+
+        litellm.logging_callback_manager.add_litellm_input_callback(my_sync_cb)
+        assert my_sync_cb in litellm.input_callback
+        assert my_sync_cb not in litellm._async_input_callback