replace with threading

Author: shahules786

nbs/llm/llm.ipynb

@@ -27,13 +27,14 @@
     "import typing as t\n",
     "import asyncio\n",
     "import inspect\n",
+    "import threading\n",
     "from pydantic import BaseModel\n",
     "import instructor\n",
     "\n",
     "T = t.TypeVar('T', bound=BaseModel)\n",
     "\n",
     "class RagasLLM:\n",
-    "    def __init__(self, provider: str, model:str, client: t.Any, **model_args):\n",
+    "    def __init__(self, provider: str, model: str, client: t.Any, **model_args):\n",
     "        self.provider = provider.lower()\n",
     "        self.model = model\n",
     "        self.model_args = model_args or {}\n",
@@ -70,23 +71,54 @@
     "    def _run_async_in_current_loop(self, coro):\n",
     "        \"\"\"Run an async coroutine in the current event loop if possible.\n",
     "        \n",
-    "        This handles Jupyter environments correctly by using the existing loop.\n",
+    "        This handles Jupyter environments correctly by using a separate thread\n",
+    "        when a running event loop is detected.\n",
     "        \"\"\"\n",
     "        try:\n",
-    "            # Check if we're in an environment with an existing event loop (like Jupyter)\n",
+    "            # Try to get the current event loop\n",
     "            loop = asyncio.get_event_loop()\n",
+    "            \n",
     "            if loop.is_running():\n",
-    "                # We're likely in a Jupyter environment\n",
-    "                import nest_asyncio\n",
-    "                nest_asyncio.apply()\n",
-    "            return loop.run_until_complete(coro)\n",
+    "                # If the loop is already running (like in Jupyter notebooks),\n",
+    "                # we run the coroutine in a separate thread with its own event loop\n",
+    "                result_container = {'result': None, 'exception': None}\n",
+    "                \n",
+    "                def run_in_thread():\n",
+    "                    # Create a new event loop for this thread\n",
+    "                    new_loop = asyncio.new_event_loop()\n",
+    "                    asyncio.set_event_loop(new_loop)\n",
+    "                    try:\n",
+    "                        # Run the coroutine in this thread's event loop\n",
+    "                        result_container['result'] = new_loop.run_until_complete(coro)\n",
+    "                    except Exception as e:\n",
+    "                        # Capture any exceptions to re-raise in the main thread\n",
+    "                        result_container['exception'] = e\n",
+    "                    finally:\n",
+    "                        # Clean up the event loop\n",
+    "                        new_loop.close()\n",
+    "                \n",
+    "                # Start the thread and wait for it to complete\n",
+    "                thread = threading.Thread(target=run_in_thread)\n",
+    "                thread.start()\n",
+    "                thread.join()\n",
+    "                \n",
+    "                # Re-raise any exceptions that occurred in the thread\n",
+    "                if result_container['exception']:\n",
+    "                    raise result_container['exception']\n",
+    "                    \n",
+    "                return result_container['result']\n",
+    "            else:\n",
+    "                # Standard case - event loop exists but isn't running\n",
+    "                return loop.run_until_complete(coro)\n",
+    "                \n",
     "        except RuntimeError:\n",
     "            # If we get a runtime error about no event loop, create a new one\n",
     "            loop = asyncio.new_event_loop()\n",
     "            asyncio.set_event_loop(loop)\n",
     "            try:\n",
     "                return loop.run_until_complete(coro)\n",
     "            finally:\n",
+    "                # Clean up\n",
     "                loop.close()\n",
     "                asyncio.set_event_loop(None)\n",
     "    \n",
@@ -129,7 +161,7 @@
     "            **self.model_args,\n",
     "        )\n",
     "\n",
-    "def ragas_llm(provider: str,model:str, client: t.Any, **model_args) -> RagasLLM:\n",
+    "def ragas_llm(provider: str, model: str, client: t.Any, **model_args) -> RagasLLM:\n",
     "    return RagasLLM(provider=provider, client=client, model=model, **model_args)"
    ]
   },

ragas_annotator/llm/llm.py

@@ -7,13 +7,14 @@
 import typing as t
 import asyncio
 import inspect
+import threading
 from pydantic import BaseModel
 import instructor
 
 T = t.TypeVar('T', bound=BaseModel)
 
 class RagasLLM:
-    def __init__(self, provider: str, model:str, client: t.Any, **model_args):
+    def __init__(self, provider: str, model: str, client: t.Any, **model_args):
         self.provider = provider.lower()
         self.model = model
         self.model_args = model_args or {}
@@ -50,23 +51,54 @@ def _initialize_client(self, provider: str, client: t.Any) -> t.Any:
     def _run_async_in_current_loop(self, coro):
         """Run an async coroutine in the current event loop if possible.
         
-        This handles Jupyter environments correctly by using the existing loop.
+        This handles Jupyter environments correctly by using a separate thread
+        when a running event loop is detected.
         """
         try:
-            # Check if we're in an environment with an existing event loop (like Jupyter)
+            # Try to get the current event loop
             loop = asyncio.get_event_loop()
+            
             if loop.is_running():
-                # We're likely in a Jupyter environment
-                import nest_asyncio
-                nest_asyncio.apply()
-            return loop.run_until_complete(coro)
+                # If the loop is already running (like in Jupyter notebooks),
+                # we run the coroutine in a separate thread with its own event loop
+                result_container = {'result': None, 'exception': None}
+                
+                def run_in_thread():
+                    # Create a new event loop for this thread
+                    new_loop = asyncio.new_event_loop()
+                    asyncio.set_event_loop(new_loop)
+                    try:
+                        # Run the coroutine in this thread's event loop
+                        result_container['result'] = new_loop.run_until_complete(coro)
+                    except Exception as e:
+                        # Capture any exceptions to re-raise in the main thread
+                        result_container['exception'] = e
+                    finally:
+                        # Clean up the event loop
+                        new_loop.close()
+                
+                # Start the thread and wait for it to complete
+                thread = threading.Thread(target=run_in_thread)
+                thread.start()
+                thread.join()
+                
+                # Re-raise any exceptions that occurred in the thread
+                if result_container['exception']:
+                    raise result_container['exception']
+                    
+                return result_container['result']
+            else:
+                # Standard case - event loop exists but isn't running
+                return loop.run_until_complete(coro)
+                
         except RuntimeError:
             # If we get a runtime error about no event loop, create a new one
             loop = asyncio.new_event_loop()
             asyncio.set_event_loop(loop)
             try:
                 return loop.run_until_complete(coro)
             finally:
+                # Clean up
                 loop.close()
                 asyncio.set_event_loop(None)
 
@@ -109,5 +141,5 @@ async def agenerate(self, prompt: str, response_model: t.Type[T]) -> T:
             **self.model_args,
         )
 
-def ragas_llm(provider: str,model:str, client: t.Any, **model_args) -> RagasLLM:
+def ragas_llm(provider: str, model: str, client: t.Any, **model_args) -> RagasLLM:
     return RagasLLM(provider=provider, client=client, model=model, **model_args)