dosubot[bot]
commented
1 month ago Hey, @edgekid! I'm here to assist you with any bugs, questions, or contributions you may have. Let's dive in and tackle this issue together!
To resolve the "asyncio.exceptions.CancelledError" error when running your script using the Ragas library, you can modify the Runner class to properly handle the cancellation of tasks. Specifically, ensure that the event loop is properly closed and that any pending tasks are cancelled.
Here is a modified version of the Runner class with added handling for task cancellation:
class Runner(threading.Thread):
def __init__(
self,
jobs: t.List[t.Tuple[t.Coroutine, str]],
desc: str,
keep_progress_bar: bool = True,
raise_exceptions: bool = True,
run_config: t.Optional[RunConfig] = None,
):
super().__init__()
self.jobs = jobs
self.desc = desc
self.keep_progress_bar = keep_progress_bar
self.raise_exceptions = raise_exceptions
self.run_config = run_config or RunConfig()
# create task
self.loop = asyncio.new_event_loop()
self.futures = as_completed(
loop=self.loop,
coros=[coro for coro, _ in self.jobs],
max_workers=self.run_config.max_workers,
)
async def _aresults(self) -> t.List[t.Any]:
results = []
for future in tqdm(
self.futures,
desc=self.desc,
total=len(self.jobs),
leave=self.keep_progress_bar,
):
r = (-1, np.nan)
try:
r = await future
except MaxRetriesExceeded as e:
logger.warning(f"max retries exceeded for {e.evolution}")
except asyncio.CancelledError:
logger.warning("Task was cancelled")
except Exception as e:
if self.raise_exceptions:
raise e
else:
logger.error(
"Runner in Executor raised an exception", exc_info=True
)
results.append(r)
return results
def run(self):
results = []
try:
results = self.loop.run_until_complete(self._aresults())
except asyncio.CancelledError:
logger.warning("Event loop was cancelled")
finally:
self.results = results
self.loop.run_until_complete(self.loop.shutdown_asyncgens())
self.loop.close()In this updated version, the asyncio.CancelledError is caught and logged in the _aresults method. Additionally, in the run method, the event loop is properly closed by calling self.loop.shutdown_asyncgens() and self.loop.close() in the finally block. This ensures that any pending tasks are properly cancelled and the event loop is cleanly shut down [1].
Alternatively, you can use the run_async_tasks function provided in the Ragas library's async_utils.py to handle running async tasks with proper error handling and optional progress bar support:
from ragas.async_utils import run_async_tasks
tasks = [coro for coro, _ in jobs]
results = run_async_tasks(tasks, show_progress=True, progress_bar_desc="Running async tasks")This function will handle the event loop and task execution, including proper handling of asyncio.CancelledError and other exceptions [2].
Additionally, ensure that you have proper rate limiting and retry mechanisms in place when interacting with the OpenAI API to avoid HTTPStatusError: Client error '429 Too Many Requests'. You can configure these settings using the RunConfig class:
@dataclass
class RunConfig:
timeout: int = 60
max_retries: int = 10
max_wait: int = 60
max_workers: int = 16
thread_timeout: float = 80.0
exception_types: t.Union[
t.Type[BaseException],
t.Tuple[t.Type[BaseException], ...],
] = (Exception,)
log_tenacity: bool = FalseEnsure that the timeout and max_retries attributes are set appropriately to handle rate limiting issues [3][4][5].
To continue talking to Dosu, mention @dosu.
jyotsanaSingh-git
jjmachan
[ ] I have checked the documentation and related resources and couldn't resolve my bug.
Describe the bug Getting "asyncio.exceptions.CancelledError" as an error
Ragas version: Python version: 3.10
Code to Reproduce
Error trace
Expected behavior Expected to output the faithfulness score without any errors.
Additional context Add any other context about the problem here.