merge

Author: aditya1503

src/cleanlab_codex/validator.py

@@ -149,6 +149,39 @@ def __init__(
             error_msg = f"Found thresholds for non-existent evaluation metrics: {_extra_thresholds}"
             raise ValueError(error_msg)
 
+    async def validate_async(
+        self,
+        query: str,
+        context: str,
+        response: str,
+        prompt: Optional[str] = None,
+        form_prompt: Optional[Callable[[str, str], str]] = None,
+    ) -> dict[str, Any]:
+        """Async version of validate"""
+        expert_task = asyncio.create_task(self.remediate_async(query))
+        detect_task = asyncio.get_running_loop().run_in_executor(
+            None,
+            self.detect,
+            query, context, response, prompt, form_prompt
+        )
+
+        # Use gather to run tasks concurrently
+        expert_result, detect_result = await asyncio.gather(expert_task, detect_task)
+        expert_answer, maybe_entry = expert_result
+        scores, is_bad_response = detect_result
+
+        # Rest of your existing logic
+        if is_bad_response and not expert_answer:
+            self._project._sdk_client.projects.entries.add_question(
+                self._project._id, question=query,
+            ).model_dump()
+
+        return {
+            "expert_answer": expert_answer if is_bad_response else None,
+            "is_bad_response": is_bad_response,
+            **scores,
+        }
+
     def validate(
         self,
         query: str,
@@ -174,30 +207,18 @@ def validate(
                 - Additional keys from a [`ThresholdedTrustworthyRAGScore`](/cleanlab_codex/types/validator/#class-thresholdedtrustworthyragscore) dictionary: each corresponds to a [TrustworthyRAG](/tlm/api/python/utils.rag/#class-trustworthyrag) evaluation metric, and points to the score for this evaluation as well as a boolean `is_bad` flagging whether the score falls below the corresponding threshold.
         """
         try:
+            # Try to use existing event loop
             loop = asyncio.get_running_loop()
-        except RuntimeError:  # No running loop
-            loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(loop)
-        expert_task = loop.create_task(self.remediate_async(query))
-        detect_task = loop.run_in_executor(None, self.detect, query, context, response, prompt, form_prompt)
-        expert_answer, maybe_entry = loop.run_until_complete(expert_task)
-        scores, is_bad_response = loop.run_until_complete(detect_task)
-        if not loop.is_running():
-            loop.close()
-        if is_bad_response:
-            if expert_answer == None:
-                # TODO: Make this async as well in the future (only if add_question takes nontrivial amt of time on the client)
-                self._project._sdk_client.projects.entries.add_question(
-                    self._project._id, question=query,
-                ).model_dump()
-        else:
-            expert_answer = None
-
-        return {
-            "expert_answer": expert_answer,
-            "is_bad_response": is_bad_response,
-            **scores,
-        }
+            future = asyncio.run_coroutine_threadsafe(
+                self.validate_async(query, context, response, prompt, form_prompt),
+                loop
+            )
+            return future.result()
+        except RuntimeError:
+            # No existing loop - create new one
+            return asyncio.run(
+                self.validate_async(query, context, response, prompt, form_prompt)
+            )
 
     def detect(
         self,