refactor: simplify apply_nest_asyncio and ensure it is applied for the Executor

- nest_asyncio has internal checks for whether it is applied, rely on those
- Ensured apply_nest_asyncio is called in the Executor class to handle nested event loops.

Author: alex-graber-pmi

ragas/src/ragas/async_utils.py

@@ -9,22 +9,6 @@
 logger = logging.getLogger(__name__)
 
 
-def apply_nest_asyncio():
-    NEST_ASYNCIO_APPLIED: bool = False
-    if is_event_loop_running():
-        # an event loop is running so call nested_asyncio to fix this
-        try:
-            import nest_asyncio
-        except ImportError:
-            raise ImportError(
-                "It seems like your running this in a jupyter-like environment. Please install nest_asyncio with `pip install nest_asyncio` to make it work."
-            )
-
-        if not NEST_ASYNCIO_APPLIED:
-            nest_asyncio.apply()
-            NEST_ASYNCIO_APPLIED = True
-
-
 def is_event_loop_running() -> bool:
     """
     Check if an event loop is currently running.
@@ -37,6 +21,20 @@ def is_event_loop_running() -> bool:
         return loop.is_running()
 
 
+def apply_nest_asyncio():
+    """Apply nest_asyncio if an event loop is running."""
+    if is_event_loop_running():
+        # an event loop is running so call nested_asyncio to fix this
+        try:
+            import nest_asyncio
+        except ImportError:
+            raise ImportError(
+                "It seems like your running this in a jupyter-like environment. Please install nest_asyncio with `pip install nest_asyncio` to make it work."
+            )
+
+        nest_asyncio.apply()
+
+
 def as_completed(
     coroutines: t.Sequence[t.Coroutine], max_workers: int = -1
 ) -> t.Iterator[asyncio.Future]:

ragas/src/ragas/executor.py

@@ -7,7 +7,7 @@
 import numpy as np
 from tqdm.auto import tqdm
 
-from ragas.async_utils import as_completed, process_futures, run
+from ragas.async_utils import apply_nest_asyncio, as_completed, process_futures, run
 from ragas.run_config import RunConfig
 from ragas.utils import ProgressBarManager, batched
 
@@ -204,6 +204,7 @@ def results(self) -> t.List[t.Any]:
         async def _async_wrapper():
             return await self.aresults()
 
+        apply_nest_asyncio()
         return run(_async_wrapper)