[DRAFT] update evaluate to be concurrent #1340
Conversation
| if not hasattr(self, '_evaluator_executor'): | ||
| self._evaluator_executor = cf.ThreadPoolExecutor(max_workers=4) |
There was a problem hiding this comment.
I don't like doing this, but I couldn't figure out another way to pass an executor that doesn't trigger a raise RuntimeError('cannot schedule new futures after shutdown') from _log_evaluation_feedback. I also don't think I ever delete this executor, but I am not 100% sure where to do so
| if example.attachments is not None: | ||
| for attachment in example.attachments: | ||
| reader = example.attachments[attachment]["reader"] | ||
| reader.seek(0) |
There was a problem hiding this comment.
This no longer works because the evaluators run in parallel. I do not know the ideal solution to this, am open to any and all ideas.
| @@ -796,15 +855,17 @@ async def _arun_evaluators( | |||
| run = current_results["run"] | |||
| example = current_results["example"] | |||
| eval_results = current_results["evaluation_results"] | |||
| for evaluator in evaluators: | |||
| lock = asyncio.Lock() | |||
There was a problem hiding this comment.
I don't think it's necessary, can remove if wanted
There was a problem hiding this comment.
if not necessary remove -- is this used only for .extend()? (I believe .extend is atomic)
https://docs.python.org/3/faq/library.html#what-kinds-of-global-value-mutation-are-thread-safe
| manager = await manager.awith_predictions_and_evaluators( | ||
| cast(ATARGET_T, target), evaluators, max_concurrency=max_concurrency | ||
| ) |
There was a problem hiding this comment.
summary evaluators are still evaluated after all the predictions and evaluations have been made, can change in the future but I think much less of a bottle neck.
No description provided.