Add retries on batch eval (3 by default) and allow for custom evaluators

Author: rogeriochaves

evaluators/ragas/langevals_ragas/lib/common.py

@@ -90,9 +90,9 @@ class _GenericEvaluatorEntry(EvaluatorEntry):
 
 
 class RagasEvaluator(BaseEvaluator[TEntry, TSettings, TResult]):
-    def _evaluate_entry(self, entry):
+    def _evaluate_entry(self, *args, **kwargs):
         disable_tqdm()
-        return super()._evaluate_entry(entry)
+        return super()._evaluate_entry(*args, **kwargs)
 
     def evaluate_batch(self, *args, **kwargs):
         restore_tqdm()

langevals/utils.py

@@ -1,6 +1,7 @@
 import importlib
 import importlib.metadata
 import pkgutil
+import re
 import textwrap
 from typing import Optional, Type, get_args
 
@@ -64,8 +65,15 @@ def get_evaluator_definitions(evaluator_cls: BaseEvaluator):
     entry_type = get_args(fields["entry"].annotation)[0]
     result_type = get_args(fields["result"].annotation)[0]
 
-    module_name, evaluator_name = evaluator_cls.__module__.split(".", 1)
-    module_name = module_name.split("langevals_")[1]
+    namespaces = evaluator_cls.__module__.split(".", 1)
+    if len(namespaces) == 2:
+        module_name, evaluator_name = namespaces
+        module_name = module_name.split("langevals_")[1]
+    else:
+        module_name = ""
+        evaluator_name = evaluator_cls.__class__.__name__
+        # CamelCase to snake_case
+        evaluator_name = re.sub(r"(?<!^)(?=[A-Z])", "_", evaluator_name).lower()
 
     if getattr(evaluator_cls, "name", None) is None:
         raise ValueError(f"Missing name attribute in {evaluator_cls}")

langevals_core/langevals_core/base_evaluator.py

@@ -19,6 +19,7 @@
 
 from pydantic import BaseModel, ConfigDict, Field
 import pandas as pd
+from tenacity import Retrying, retry, stop_after_attempt, wait_exponential
 from tqdm.auto import tqdm
 from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, as_completed, wait
 from langevals_core.azure_patch import patch_litellm
@@ -267,9 +268,10 @@ def set_model_envs(self):
     def evaluate(self, entry: TEntry) -> SingleEvaluationResult:
         raise NotImplementedError("This method should be implemented by subclasses.")
 
-    def _evaluate_entry(self, entry):
+    def _evaluate_entry(self, entry, retries=0):
         try:
-            return self.evaluate(entry)
+            retryer = Retrying(stop=stop_after_attempt(retries), reraise=True)
+            return retryer(self.evaluate, entry)
         except Exception as exception:
             return EvaluationResultError(
                 error_type=type(exception).__name__,
@@ -284,14 +286,15 @@ def evaluate_batch(
         data: List[TEntry],
         index=0,
         max_evaluations_in_parallel=50,
+        retries=3,
         _executor_ref: Optional[Callable[[ThreadPoolExecutor], None]] = None,
     ) -> BatchEvaluationResult:
         results: list[SingleEvaluationResult] = [
             EvaluationResultSkipped(details="not processed")
         ] * len(data)
         with ThreadPoolExecutor(max_workers=max_evaluations_in_parallel) as executor:
             future_to_index = {
-                executor.submit(self._evaluate_entry, entry): idx
+                executor.submit(self._evaluate_entry, entry, retries): idx
                 for idx, entry in enumerate(data)
             }
 
@@ -306,7 +309,9 @@ def evaluate_batch(
                             executor, "interrupted"
                         ) and executor.__getattribute__("interrupted"):
                             raise KeyboardInterrupt()
-                        done, not_done = wait(not_done, timeout=0.1, return_when=FIRST_COMPLETED)
+                        done, not_done = wait(
+                            not_done, timeout=0.1, return_when=FIRST_COMPLETED
+                        )
                         for future in done:
                             idx = future_to_index[future]
                             results[idx] = future.result()