fix(executor): simplify process_futures calls and progress bar updates

- Removed optional progress bar parameter from process_futures.
- Updated calls to process_futures to handle progress bar updates directly.
- Cleaned up error handling logic for results in async processing.

Author: alex-graber-pmi

ragas/src/ragas/async_utils.py

@@ -4,8 +4,6 @@
 import logging
 import typing as t
 
-from tqdm.auto import tqdm
-
 logger = logging.getLogger(__name__)
 
 
@@ -58,14 +56,13 @@ async def sema_coro(coro):
 
 
 async def process_futures(
-    futures: t.Iterator[asyncio.Future], pbar: t.Optional[tqdm] = None
+    futures: t.Iterator[asyncio.Future],
 ) -> t.AsyncGenerator[t.Any, None]:
     """
     Process futures with optional progress tracking.
 
     Args:
         futures: Iterator of asyncio futures to process (e.g., from asyncio.as_completed)
-        pbar: Optional progress bar to update
 
     Yields:
         Results from completed futures as they finish
@@ -77,8 +74,6 @@ async def process_futures(
         except Exception as e:
             result = e
 
-        if pbar:
-            pbar.update(1)
         yield result
 
 
@@ -139,10 +134,9 @@ async def _run():
 
         if not batch_size:
             with pbm.create_single_bar(total_tasks) as pbar:
-                async for result in process_futures(
-                    as_completed(tasks, max_workers), pbar
-                ):
+                async for result in process_futures(as_completed(tasks, max_workers)):
                     results.append(result)
+                    pbar.update(1)
         else:
             total_tasks = len(tasks)
             batches = batched(tasks, batch_size)
@@ -153,9 +147,10 @@ async def _run():
                 for i, batch in enumerate(batches, 1):
                     pbm.update_batch_bar(batch_pbar, i, n_batches, len(batch))
                     async for result in process_futures(
-                        as_completed(batch, max_workers), batch_pbar
+                        as_completed(batch, max_workers)
                     ):
                         results.append(result)
+                        batch_pbar.update(1)
                     overall_pbar.update(len(batch))
 
         return results

ragas/src/ragas/executor.py

@@ -148,36 +148,19 @@ async def _process_batched_jobs(
                     afunc(*args, **kwargs) for afunc, args, kwargs, _ in batch
                 ]
                 async for result in process_futures(
-                    as_completed(coroutines, max_workers), batch_pbar
+                    as_completed(coroutines, max_workers)
                 ):
-                    # Ensure result is always a tuple (counter, value)
-                    if isinstance(result, Exception):
-                        # Find the counter for this failed job
-                        idx = coroutines.index(result.__context__)
-                        counter = (
-                            batch[idx][0].__closure__[1].cell_contents
-                        )  # counter from closure
-                        results.append((counter, result))
-                    else:
-                        results.append(result)
+                    results.append(result)
+                    batch_pbar.update(1)
                 # Update overall progress bar for all futures in this batch
                 overall_pbar.update(len(batch))
 
     async def _process_coroutines(self, jobs, pbar, results, max_workers):
         """Helper function to process coroutines and update the progress bar."""
         coroutines = [afunc(*args, **kwargs) for afunc, args, kwargs, _ in jobs]
-        async for result in process_futures(
-            as_completed(coroutines, max_workers), pbar
-        ):
-            # Ensure result is always a tuple (counter, value)
-            if isinstance(result, Exception):
-                idx = coroutines.index(result.__context__)
-                counter = (
-                    jobs[idx][0].__closure__[1].cell_contents
-                )  # counter from closure
-                results.append((counter, result))
-            else:
-                results.append(result)
+        async for result in process_futures(as_completed(coroutines, max_workers)):
+            results.append(result)
+            pbar.update(1)
 
     async def aresults(self) -> t.List[t.Any]:
         """