Colors on print messages, hanging threads, allow no type in file, cancel run on program abort

Author: Andrei Bratu

src/humanloop/eval_utils/run.py

@@ -19,7 +19,7 @@
 import time
 import types
 import typing
-from concurrent.futures import ThreadPoolExecutor
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime
 from functools import partial
 from logging import INFO
@@ -204,7 +204,10 @@ def _overload_log(
             response = self._log(**kwargs)
         except Exception as e:
             error_message = str(e).replace("\n", " ")
-            sys.stderr.write(f"{RED}Failed to log: {error_message[:100]}...{RESET}\n")
+            if len(error_message) > 100:
+                sys.stderr.write(f"{RED}Failed to log: {error_message[:100]}...{RESET}\n")
+            else:
+                sys.stderr.write(f"{RED}Failed to log: {error_message}{RESET}\n")
             raise e
 
         # Notify the run_eval utility about one Log being created
@@ -362,13 +365,25 @@ def upload_callback(log_id: str):
                     end_time=datetime.now(),
                 )
                 error_message = str(e).replace("\n", " ")
-                sys.stderr.write(
-                    f"\n{RED}Your {hl_file.type}'s `callable` failed for Datapoint: {dp.id}. Error: {error_message[:100]}...{RESET}\n"
-                )
+                if len(error_message) > 100:
+                    sys.stderr.write(
+                        f"\n{RED}Your {hl_file.type}'s `callable` failed for Datapoint: {dp.id}. Error: {error_message[:100]}...{RESET}\n"
+                    )
+                else:
+                    sys.stderr.write(
+                        f"\n{RED}Your {hl_file.type}'s `callable` failed for Datapoint: {dp.id}. Error: {error_message}{RESET}\n"
+                    )
 
         with ThreadPoolExecutor(max_workers=workers) as executor:
+            futures = []
             for datapoint in hl_dataset.datapoints:
-                executor.submit(_process_datapoint, datapoint)
+                futures.append(executor.submit(_process_datapoint, datapoint))
+            # Program hangs if any uncaught exceptions are not handled here
+            for future in as_completed(futures):
+                try:
+                    future.result()
+                except Exception:
+                    pass
 
     stats = _wait_for_evaluation_to_complete(
         client=client,
@@ -901,13 +916,8 @@ def _run_local_evaluators(
     progress_bar: _SimpleProgressBar,
 ):
     """Run local Evaluators on the Log and send the judgments to Humanloop."""
-    # If there are no local evaluators, we don't need to do the log lookup.
-    if len(local_evaluators) == 0:
-        progress_bar.increment()
-        return
-
+    # Need to get the full log to pass to the evaluators
     try:
-        # Need to get the full log to pass to the evaluators
         log = client.logs.get(id=log_id)
         if not isinstance(log, dict):
             log_dict = log.dict()
@@ -921,6 +931,7 @@ def _run_local_evaluators(
                 log_dict = log.dict()
             else:
                 log_dict = log
+            time.sleep(2)
         datapoint_dict = datapoint.dict() if datapoint else None
 
         for local_evaluator_tuple in local_evaluators:
@@ -953,14 +964,24 @@ def _run_local_evaluators(
                     end_time=datetime.now(),
                 )
                 error_message = str(e).replace("\n", " ")
-                sys.stderr.write(
-                    f"{RED}Evaluator {local_evaluator.path} failed with error {error_message[:100]}...{RESET}\n"
-                )
+                if len(error_message) > 100:
+                    sys.stderr.write(
+                        f"{RED}Evaluator {local_evaluator.path} failed with error {error_message[:100]}...{RESET}\n"
+                    )
+                else:
+                    sys.stderr.write(
+                        f"{RED}Evaluator {local_evaluator.path} failed with error {error_message}{RESET}\n"
+                    )
     except Exception as e:
         error_message = str(e).replace("\n", " ")
-        sys.stderr.write(
-            f"{RED}Failed to run local Evaluators for source datapoint {datapoint.dict()['id'] if datapoint else None}: {error_message[:100]}...{RESET}\n"
-        )
+        if len(error_message) > 100:
+            sys.stderr.write(
+                f"{RED}Failed to run local Evaluators for source datapoint {datapoint.dict()['id'] if datapoint else None}: {error_message[:100]}...{RESET}\n"
+            )
+        else:
+            sys.stderr.write(
+                f"{RED}Failed to run local Evaluators for source datapoint {datapoint.dict()['id'] if datapoint else None}: {error_message}{RESET}\n"
+            )
         pass
     finally:
         progress_bar.increment()