Retry an async operation :param tasks: :return:
(self, tasks)
| 75 | return result |
| 76 | |
| 77 | async def retry_operation(self, tasks): |
| 78 | """ |
| 79 | Retry an async operation |
| 80 | :param tasks: |
| 81 | :return: |
| 82 | """ |
| 83 | delay = self.llm_config.async_params.retry_interval |
| 84 | timeout = delay * self.llm_config.async_params.max_retries |
| 85 | |
| 86 | start_time = asyncio.get_event_loop().time() |
| 87 | end_time = start_time + timeout |
| 88 | results = [] |
| 89 | while True: |
| 90 | remaining_time = end_time - asyncio.get_event_loop().time() |
| 91 | if remaining_time <= 0: |
| 92 | print("Timeout reached. Operation incomplete.") |
| 93 | break |
| 94 | |
| 95 | done, pending = await asyncio.wait(tasks, timeout=delay) |
| 96 | results += list(done) |
| 97 | |
| 98 | if len(done) == len(tasks): |
| 99 | print("All tasks completed successfully.") |
| 100 | break |
| 101 | |
| 102 | if not pending: |
| 103 | print("No pending tasks. Operation incomplete.") |
| 104 | break |
| 105 | |
| 106 | tasks = list(pending) # Retry with the pending tasks |
| 107 | return results |
| 108 | |
| 109 | async def async_batch_invoke(self, inputs: list[dict]) -> list[dict]: |
| 110 | """ |